Step-by-Step CDC Recovery Guide: Oracle to PostgreSQL

A runbook for keeping AWS DMS change data capture recoverable between a self-managed Oracle source and a PostgreSQL target on AWS, and for recovering it when a task fails. It is a worked example from one migration: schema names, table names and SCNs are illustrative.

Everything here rests on one constraint. DMS can only resume from a point that still exists in the source database’s transaction logs. Every configuration step below exists to widen that window or to record where the useful points in it are.

Recovery points and their limits

A CDC start position is a native position in the source’s transaction log — an SCN for Oracle, an LSN for PostgreSQL, a binlog file and offset for MySQL.

A CDC start time is a timestamp, which DMS converts to the corresponding native position. PostgreSQL as a source does not support this, because there is no mapping from a timestamp to an LSN.

The important constraint on both: a task’s CDC start point is fixed when the task is created and cannot be changed. AWS’s documentation is explicit — to use a different CDC start point, create a new task. Recovery procedures built on modify-replication-task --cdc-start-position against an existing task do not do what they appear to do; the procedures below create a new CDC-only task instead.

What limits how far back you can go:

Log retention. Oracle keeps redo and archived redo logs; PostgreSQL keeps WAL segments; MySQL keeps binary logs. Each is reclaimed on its own schedule.
Log availability. Once logs are purged, the position is unreachable and the only route back is a fresh full load followed by CDC.
Engine constraints. Oracle needs ARCHIVELOG mode and adequate retention; PostgreSQL needs its logical replication slot maintained; MySQL’s binlog_expire_logs_seconds sets the ceiling.

Three habits make recovery possible rather than theoretical: extend log retention deliberately rather than accepting the default, record the source’s current position on a schedule so there is a known-good point to recover to, and alert on log space and purging rather than discovering the problem during an incident.

Part 1: Configuration

Step 1: Oracle source

Enable ARCHIVELOG mode, if it is not already:

SHUTDOWN IMMEDIATE;
STARTUP MOUNT;
ALTER DATABASE ARCHIVELOG;
ALTER DATABASE OPEN;

SELECT log_mode FROM v$database;

Configure supplemental logging so that changes carry enough information to be applied elsewhere:

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;

-- Tables without a primary key need all columns logged
ALTER TABLE schema.table_without_pk ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

Retain archived redo logs for the intended recovery window. Two settings govern this: the size of the fast recovery area, and the RMAN deletion policy that decides when a log inside it may be reclaimed.

-- How much space the fast recovery area may use
SELECT name, value FROM v$parameter WHERE name = 'db_recovery_file_dest_size';
ALTER SYSTEM SET db_recovery_file_dest_size = 100G SCOPE=BOTH;

-- In RMAN: do not reclaim an archived log until it has been backed up
CONFIGURE ARCHIVELOG DELETION POLICY TO BACKED UP 1 TIMES TO DISK;

Size the recovery area from the measured redo generation rate (the monitoring view below gives it) multiplied by the window you want, with headroom — under-sizing it causes logs to be reclaimed early, which is the failure this whole section exists to avoid.

DB_FLASHBACK_RETENTION_TARGET is sometimes cited here and does not belong. It is expressed in minutes, not hours — the default of 1440 is 24 hours, so a value of 168 is two hours and 48 minutes, not seven days — and it governs how far back Flashback Database can rewind the database using flashback logs in the fast recovery area. It has no effect on how long archived redo logs are kept, and therefore none on the CDC recovery window.

Record the current SCN on a schedule, so there is always a recent known-good position:

CREATE TABLE admin.dms_scn_checkpoints (
  checkpoint_id   NUMBER GENERATED ALWAYS AS IDENTITY,
  checkpoint_time TIMESTAMP DEFAULT SYSTIMESTAMP,
  checkpoint_name VARCHAR2(100),
  scn             NUMBER,
  notes           VARCHAR2(4000),
  PRIMARY KEY (checkpoint_id)
);

CREATE OR REPLACE PROCEDURE admin.capture_scn_checkpoint(
  p_checkpoint_name IN VARCHAR2,
  p_notes           IN VARCHAR2 DEFAULT NULL
)
AS
  v_current_scn NUMBER;
BEGIN
  SELECT CURRENT_SCN INTO v_current_scn FROM V$DATABASE;

  INSERT INTO admin.dms_scn_checkpoints (checkpoint_name, scn, notes)
  VALUES (p_checkpoint_name, v_current_scn, p_notes);

  COMMIT;
END;
/

BEGIN
  DBMS_SCHEDULER.CREATE_JOB (
    job_name        => 'SCN_CHECKPOINT_JOB',
    job_type        => 'STORED_PROCEDURE',
    job_action      => 'admin.capture_scn_checkpoint',
    start_date      => SYSTIMESTAMP,
    repeat_interval => 'FREQ=HOURLY;INTERVAL=6',
    enabled         => TRUE,
    comments        => 'Capture an SCN checkpoint every 6 hours',
    auto_drop       => FALSE,
    job_class       => 'DEFAULT_JOB_CLASS'
  );

  DBMS_SCHEDULER.SET_JOB_ARGUMENT_VALUE (
    job_name          => 'SCN_CHECKPOINT_JOB',
    argument_position => 1,
    argument_value    => 'SCHEDULED_CHECKPOINT'
  );
END;
/

Also capture a checkpoint by hand before and after any large data load or schema change — those are the points a recovery is most likely to need.

Monitor archive log generation, which is both the input to sizing and the early warning that the window is shrinking:

CREATE OR REPLACE VIEW admin.archive_log_stats AS
SELECT
  TRUNC(FIRST_TIME) AS log_date,
  COUNT(*) AS logs_generated,
  ROUND(SUM(BLOCKS * BLOCK_SIZE) / 1024 / 1024 / 1024, 2) AS size_gb
FROM V$ARCHIVED_LOG
WHERE FIRST_TIME > SYSDATE - 7
GROUP BY TRUNC(FIRST_TIME)
ORDER BY log_date;

Step 2: PostgreSQL target

A table to record, on the target side, which source position each recovery point corresponds to:

CREATE SCHEMA IF NOT EXISTS admin;

CREATE TABLE admin.dms_cdc_checkpoints (
  checkpoint_id   SERIAL PRIMARY KEY,
  checkpoint_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  checkpoint_name VARCHAR(100),
  task_arn        TEXT,
  oracle_scn      NUMERIC,
  postgres_lsn    TEXT,
  notes           TEXT
);

Settings that matter for DMS throughput and for any later use of the target as a source:

ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_wal_senders = 10;
ALTER SYSTEM SET max_replication_slots = 10;

SELECT pg_reload_conf();

wal_level needs a restart, not a reload. On Amazon RDS and Aurora these are parameter group settings rather than ALTER SYSTEM statements.

Step 3: The DMS task

aws dms create-replication-task \
  --replication-task-identifier "oracle-to-pg-migration" \
  --source-endpoint-arn "arn:aws:dms:region:account:endpoint:source-oracle-endpoint" \
  --target-endpoint-arn "arn:aws:dms:region:account:endpoint:target-pg-endpoint" \
  --replication-instance-arn "arn:aws:dms:region:account:rep:instance-name" \
  --migration-type "full-load-and-cdc" \
  --table-mappings file://table-mappings.json \
  --replication-task-settings file://task-settings.json

Task settings, with logging turned up enough to diagnose a CDC failure after the fact:

{
  "TargetMetadata": {
    "TargetSchema": "",
    "SupportLobs": true,
    "FullLobMode": false,
    "LimitedSizeLobMode": true,
    "LobChunkSize": 64,
    "LobMaxSize": 32,
    "BatchApplyEnabled": true
  },
  "FullLoadSettings": {
    "TargetTablePrepMode": "DO_NOTHING",
    "CreatePkAfterFullLoad": false,
    "StopTaskCachedChangesApplied": false,
    "StopTaskCachedChangesNotApplied": false,
    "MaxFullLoadSubTasks": 8,
    "TransactionConsistencyTimeout": 600,
    "CommitRate": 10000
  },
  "Logging": {
    "EnableLogging": true,
    "LogComponents": [
      { "Id": "TRANSFORMATION", "Severity": "LOGGER_SEVERITY_DEFAULT" },
      { "Id": "SOURCE_UNLOAD",  "Severity": "LOGGER_SEVERITY_DEFAULT" },
      { "Id": "IO",             "Severity": "LOGGER_SEVERITY_DEFAULT" },
      { "Id": "TARGET_LOAD",    "Severity": "LOGGER_SEVERITY_DEFAULT" },
      { "Id": "PERFORMANCE",    "Severity": "LOGGER_SEVERITY_DEFAULT" }
    ]
  },
  "OracleSettings": {
    "ReadTableSpaceName": false,
    "EnableHomogenousTablespace": false,
    "StandbyDelayTime": 0,
    "ArchivedLogsOnly": false,
    "ArchivedLogDestId": 0,
    "UseLogminerReader": true,
    "SecurityDbEncryption": false,
    "DirectPathNoLog": false,
    "AllowSelectNestedTables": true,
    "ConvertTimestampWithZoneToUTC": true,
    "AddSupplementalLogging": true
  },
  "ChangeProcessingTuning": {
    "BatchApplyPreserveTransaction": true,
    "BatchApplyTimeoutMin": 1,
    "BatchApplyTimeoutMax": 30,
    "BatchApplyMemoryLimit": 500,
    "BatchSplitSize": 0,
    "MinTransactionSize": 1000,
    "CommitTimeout": 1,
    "MemoryLimitTotal": 1024,
    "MemoryKeepTime": 60,
    "StatementCacheSize": 50
  }
}

Consider also setting TaskRecoveryTableEnabled to Yes, which makes DMS write its recovery checkpoint continuously to an awsdms_txn_state table on the target. That checkpoint string can be handed straight to --cdc-start-position on a replacement task, and unlike the API’s copy it survives the original task being deleted.

Recording checkpoints. A script run hourly, correlating the DMS task’s state with the source’s current SCN. Neither database password appears in the script or in a process argument: Oracle is reached through an external password store (an Oracle Wallet alias), and the PostgreSQL password comes from Secrets Manager into PGPASSWORD for the life of the process.

#!/bin/bash
set -euo pipefail

TASK_ARN="arn:aws:dms:region:account:task:your-task-arn"
PG_HOST="postgres-host"
PG_PORT="5432"
PG_DB="postgres-db"
PG_USER="dms_monitor"
PG_SECRET_ID="prod/postgres/dms_monitor"
ORACLE_WALLET_ALIAS="DMS_MONITOR"   # tnsnames alias with wallet credentials

TASK_INFO=$(aws dms describe-replication-tasks \
  --filters Name=replication-task-arn,Values="$TASK_ARN" \
  --query 'ReplicationTasks[0]')

TASK_STATUS=$(echo "$TASK_INFO" | jq -r '.Status')
if [ "$TASK_STATUS" != "running" ]; then
  echo "Task is $TASK_STATUS, not running. Skipping checkpoint."
  exit 0
fi

CHECKPOINT_NAME="AUTO_$(date +%Y%m%d_%H%M%S)"
CDC_LATENCY=$(echo "$TASK_INFO" | jq -r '.ReplicationTaskStats.CdcLatencySource')

# Quoted heredoc: no shell expansion, so V$DATABASE needs no escaping
ORACLE_SCN=$(sqlplus -s /nolog <<'EOF' | tr -d ' '
CONNECT /@DMS_MONITOR
SET HEADING OFF FEEDBACK OFF VERIFY OFF PAGESIZE 0
SELECT CURRENT_SCN FROM V$DATABASE;
EXIT;
EOF
)

PGPASSWORD=$(aws secretsmanager get-secret-value \
  --secret-id "$PG_SECRET_ID" --query SecretString --output text \
  | python3 -c 'import json,sys; print(json.load(sys.stdin)["password"])')
export PGPASSWORD

psql -h "$PG_HOST" -p "$PG_PORT" -d "$PG_DB" -U "$PG_USER" \
     -v ON_ERROR_STOP=1 \
     -v name="$CHECKPOINT_NAME" -v arn="$TASK_ARN" \
     -v scn="$ORACLE_SCN" -v latency="$CDC_LATENCY" <<'EOF'
INSERT INTO admin.dms_cdc_checkpoints (checkpoint_name, task_arn, oracle_scn, notes)
VALUES (:'name', :'arn', :'scn'::numeric,
        'Automatic checkpoint. CDC source latency: ' || :'latency' || ' seconds');
EOF

echo "Recorded checkpoint $CHECKPOINT_NAME at SCN $ORACLE_SCN"

Scheduled hourly:

0 * * * * /path/to/dms-record-checkpoints.sh >> /var/log/dms-checkpoints.log 2>&1

Part 2: Recovery

Step 1: Find the available recovery points

Recorded checkpoints, on the Oracle side:

SELECT checkpoint_id, checkpoint_time, checkpoint_name, scn, notes
FROM admin.dms_scn_checkpoints
ORDER BY checkpoint_time DESC;

Which archived logs still exist:

SELECT sequence#, first_change#, first_time, next_change#, next_time, archived, status
FROM v$archived_log
WHERE first_time > SYSDATE - 7
ORDER BY sequence# DESC;

The oldest position still reachable — this is the hard floor on recovery:

SELECT MIN(first_change#) AS min_available_scn,
       TO_CHAR(MIN(first_time), 'YYYY-MM-DD HH24:MI:SS') AS min_available_time
FROM v$archived_log
WHERE status = 'A';

And the checkpoints recorded on the target:

SELECT checkpoint_id, checkpoint_time, checkpoint_name, task_arn, oracle_scn, notes
FROM admin.dms_cdc_checkpoints
ORDER BY checkpoint_time DESC;

Step 2: Recover from a specific SCN

Stop the failed task and confirm the SCN is still covered by an available archived log (Step 4 below), then create a new CDC-only task at that position. The start position cannot be changed on the existing task, so a new task is the procedure rather than a fallback.

aws dms stop-replication-task \
  --replication-task-arn "arn:aws:dms:region:account:task:your-task-arn"

aws dms create-replication-task \
  --replication-task-identifier "oracle-to-pg-recovery-$(date +%Y%m%d-%H%M%S)" \
  --source-endpoint-arn "arn:aws:dms:region:account:endpoint:source-oracle-endpoint" \
  --target-endpoint-arn "arn:aws:dms:region:account:endpoint:target-pg-endpoint" \
  --replication-instance-arn "arn:aws:dms:region:account:rep:instance-name" \
  --migration-type "cdc" \
  --table-mappings file://table-mappings.json \
  --replication-task-settings file://task-settings.json \
  --cdc-start-position 6916533

The value is a bare SCN. --cdc-start-position accepts a date, a checkpoint string or a native LSN/SCN, with no engine-name prefix.

aws dms start-replication-task \
  --replication-task-arn "<new-task-arn>" \
  --start-replication-task-type start-replication

Watch it catch up:

aws dms describe-replication-tasks \
  --filters Name=replication-task-arn,Values="<new-task-arn>" \
  --query 'ReplicationTasks[0].{Status:Status,SourceLatency:ReplicationTaskStats.CdcLatencySource,TargetLatency:ReplicationTaskStats.CdcLatencyTarget}'

Delete the old task only once the replacement has been verified — its recovery checkpoint is lost with it.

Two caveats specific to Oracle. Starting from an SCN misses transactions that were already open at that point and committed afterwards; from DMS 3.5.1, the openTransactionWindow endpoint setting takes a number of minutes to scan back for them. And because DMS gives at-least-once delivery, restarting from a point slightly before the failure will re-apply changes: every replicated table needs a primary key or unique index so those land as updates rather than duplicates.

Step 3: Recover from a timestamp

Convert the timestamp to an SCN and use the SCN, which removes the ambiguity of a timestamp that maps to several positions:

SELECT scn
FROM admin.dms_scn_checkpoints
WHERE checkpoint_time <= TO_TIMESTAMP('2025-02-15 14:30:00', 'YYYY-MM-DD HH24:MI:SS')
ORDER BY checkpoint_time DESC
FETCH FIRST 1 ROW ONLY;

-- Or ask Oracle directly
SELECT timestamp_to_scn(TO_TIMESTAMP('2025-02-15 14:30:00', 'YYYY-MM-DD HH24:MI:SS'))
FROM dual;

Alternatively, create the task with --cdc-start-time and let DMS do the conversion. The two options are mutually exclusive; supplying both is an error.

aws dms create-replication-task \
  --replication-task-identifier "oracle-to-pg-recovery-$(date +%Y%m%d-%H%M%S)" \
  --source-endpoint-arn "arn:aws:dms:region:account:endpoint:source-oracle-endpoint" \
  --target-endpoint-arn "arn:aws:dms:region:account:endpoint:target-pg-endpoint" \
  --replication-instance-arn "arn:aws:dms:region:account:rep:instance-name" \
  --migration-type "cdc" \
  --table-mappings file://table-mappings.json \
  --replication-task-settings file://task-settings.json \
  --cdc-start-time 2025-02-15T14:30:00Z

Check the task’s log group for a successful connection at the requested position before assuming the recovery worked:

aws logs get-log-events \
  --log-group-name /aws/dms/replication-task/<task-id> \
  --log-stream-name <task-log-stream> \
  --limit 100

Step 4: When the recovery point is unavailable

Confirm whether an archived log still spans the SCN. This single query answers the question and should be run before creating the recovery task, not after it fails:

SELECT sequence#, first_change#, next_change#, archived, status
FROM v$archived_log
WHERE first_change# <= 6916533 AND next_change# > 6916533;

If a deeper check is needed, LogMiner can read that specific log — but note that DBMS_LOGMNR.CONTINUOUS_MINE is desupported from Oracle Database 19c (19.1) with no replacement, so the log files must be added explicitly:

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE(
    LOGFILENAME => '/path/to/archivelog/1_12345_1043678901.dbf',
    OPTIONS     => DBMS_LOGMNR.NEW);

  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN => 6916533,
    OPTIONS  => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG);

  FOR rec IN (
    SELECT operation, sql_redo
    FROM v$logmnr_contents
    WHERE scn >= 6916533 AND ROWNUM <= 10
  ) LOOP
    DBMS_OUTPUT.PUT_LINE(rec.operation || ': ' || rec.sql_redo);
  END LOOP;

  DBMS_LOGMNR.END_LOGMNR;
END;
/

The log file names come from the v$archived_log query above. If none covers the SCN, the position is gone: find the nearest position that is still available,

SELECT sequence#,
       first_change# AS start_scn, first_time AS start_time,
       next_change#  AS end_scn,   next_time  AS end_time
FROM v$archived_log
WHERE status = 'A' AND archived = 'YES'
ORDER BY sequence#;

and accept that changes between the last applied position and that point were never replicated. Reconcile those tables from the source — a targeted reload of the affected window, as in the time-window reload guide, or a full reload where the gap is large.

Part 3: Post-recovery validation

Row counts across the target, and how many rows changed since the recovery point:

CREATE OR REPLACE FUNCTION admin.verify_table_counts(
  p_start_time TIMESTAMP DEFAULT NULL
) RETURNS TABLE (
  table_name TEXT,
  record_count BIGINT,
  updated_since_recovery BIGINT
) AS $$
DECLARE
  r RECORD;
  qualified TEXT;
BEGIN
  FOR r IN
    SELECT t.table_schema, t.table_name
    FROM information_schema.tables t
    WHERE t.table_schema = 'public' AND t.table_type = 'BASE TABLE'
  LOOP
    qualified := format('%I.%I', r.table_schema, r.table_name);
    table_name := qualified;

    EXECUTE format('SELECT COUNT(*) FROM %s', qualified) INTO record_count;

    updated_since_recovery := 0;
    IF p_start_time IS NOT NULL THEN
      BEGIN
        EXECUTE format('SELECT COUNT(*) FROM %s WHERE updated_at >= $1', qualified)
          INTO updated_since_recovery
          USING p_start_time;
      EXCEPTION WHEN undefined_column THEN
        updated_since_recovery := 0;   -- table has no updated_at
      END;
    END IF;

    RETURN NEXT;
  END LOOP;
END;
$$ LANGUAGE plpgsql;

SELECT * FROM admin.verify_table_counts('2025-02-15 14:30:00');

Two details in that function are load-bearing. Identifiers are interpolated with format('%I') rather than string concatenation, so a table name cannot alter the statement. And the dynamic statement is written EXECUTE … INTO … USING … in that order — PL/pgSQL rejects USING before INTO.

Watch CDC latency while the replacement task catches up:

#!/bin/bash
TASK_ARN="arn:aws:dms:region:account:task:your-task-arn"

while true; do
  STATS=$(aws dms describe-replication-tasks \
    --filters Name=replication-task-arn,Values="$TASK_ARN" \
    --query 'ReplicationTasks[0].ReplicationTaskStats')

  echo "$(date +'%F %T') source=$(echo "$STATS" | jq -r '.CdcLatencySource')s" \
       "target=$(echo "$STATS" | jq -r '.CdcLatencyTarget')s"
  sleep 60
done

Business-level checks catch what row counts miss — a recovery that re-applied changes in the wrong order leaves consistent counts and inconsistent totals:

CREATE TABLE IF NOT EXISTS admin.data_validations (
  validation_id   SERIAL PRIMARY KEY,
  validation_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  passed          BOOLEAN,
  details         TEXT
);

CREATE OR REPLACE PROCEDURE admin.validate_critical_tables()
LANGUAGE plpgsql
AS $$
DECLARE
  validation_passed  BOOLEAN := TRUE;
  total_customers    INTEGER;
  total_orders       INTEGER;
  total_order_items  INTEGER;
  validation_msg     TEXT := '';
BEGIN
  SELECT COUNT(*) INTO total_customers   FROM public.customers;
  SELECT COUNT(*) INTO total_orders      FROM public.orders;
  SELECT COUNT(*) INTO total_order_items FROM public.order_items;

  IF total_orders > 0 AND total_order_items = 0 THEN
    validation_passed := FALSE;
    validation_msg := validation_msg || 'Orders exist but no order items. ';
  END IF;

  IF total_customers = 0 THEN
    validation_passed := FALSE;
    validation_msg := validation_msg || 'No customers found. ';
  END IF;

  INSERT INTO admin.data_validations (validation_time, passed, details)
  VALUES (CURRENT_TIMESTAMP, validation_passed, NULLIF(validation_msg, ''));

  RAISE NOTICE 'Validation passed: %, message: %',
    validation_passed, COALESCE(NULLIF(validation_msg, ''), 'No issues found.');
END;
$$;

CALL admin.validate_critical_tables();

Part 4: Automating recovery

A scheduled Lambda that records task health and, when a task has failed or fallen badly behind, creates a replacement CDC task from the last recorded checkpoint.

Treat automatic recovery with caution. It creates a new DMS task on each attempt and re-applies changes from the checkpoint forward, which is safe only if the target is idempotent. Put a guard on how often it may fire, and alert on every attempt — a recovery loop that nobody is watching converts a visible outage into an invisible one.

import json
import os

import boto3
import psycopg2

LATENCY_THRESHOLD_SECONDS = 1800


def db_password(secret_id):
    secrets = boto3.client("secretsmanager")
    secret = secrets.get_secret_value(SecretId=secret_id)["SecretString"]
    return json.loads(secret)["password"]


def lambda_handler(event, context):
    task_arn = os.environ["DMS_TASK_ARN"]
    dms = boto3.client("dms")

    try:
        task = dms.describe_replication_tasks(
            Filters=[{"Name": "replication-task-arn", "Values": [task_arn]}]
        )["ReplicationTasks"][0]

        status = task["Status"]
        stats = task.get("ReplicationTaskStats", {})
        source_latency = stats.get("CdcLatencySource", 0)
        target_latency = stats.get("CdcLatencyTarget", 0)

        conn = psycopg2.connect(
            host=os.environ["PG_HOST"],
            port=os.environ["PG_PORT"],
            dbname=os.environ["PG_DB"],
            user=os.environ["PG_USER"],
            password=db_password(os.environ["PG_SECRET_ID"]),
        )
        cur = conn.cursor()

        cur.execute(
            """
            INSERT INTO admin.dms_monitoring
              (task_arn, status, source_latency, target_latency)
            VALUES (%s, %s, %s, %s)
            """,
            (task_arn, status, source_latency, target_latency),
        )

        needs_recovery = status == "failed" or source_latency > LATENCY_THRESHOLD_SECONDS
        if not needs_recovery:
            message = "CDC functioning normally"
        else:
            cur.execute(
                """
                SELECT checkpoint_name, oracle_scn
                FROM admin.dms_cdc_checkpoints
                WHERE task_arn = %s AND oracle_scn IS NOT NULL
                ORDER BY checkpoint_time DESC
                LIMIT 1
                """,
                (task_arn,),
            )
            result = cur.fetchone()

            if not result:
                message = "No valid checkpoint found for recovery"
            else:
                checkpoint_name, oracle_scn = result

                cur.execute(
                    """
                    INSERT INTO admin.dms_recovery_events
                      (task_arn, event_type, checkpoint_name, oracle_scn, notes)
                    VALUES (%s, %s, %s, %s, %s)
                    """,
                    (
                        task_arn,
                        "AUTO_RECOVERY",
                        checkpoint_name,
                        oracle_scn,
                        f"status={status} source_latency={source_latency}",
                    ),
                )

                if status != "stopped":
                    dms.stop_replication_task(ReplicationTaskArn=task_arn)

                # A task's CDC start point cannot be changed, so recovery means a
                # new CDC-only task. The failed task is left in place so that its
                # recovery checkpoint survives for manual inspection.
                new_task = dms.create_replication_task(
                    ReplicationTaskIdentifier=f"{task['ReplicationTaskIdentifier']}-r{int(oracle_scn)}",
                    SourceEndpointArn=task["SourceEndpointArn"],
                    TargetEndpointArn=task["TargetEndpointArn"],
                    ReplicationInstanceArn=task["ReplicationInstanceArn"],
                    MigrationType="cdc",
                    TableMappings=task["TableMappings"],
                    ReplicationTaskSettings=task["ReplicationTaskSettings"],
                    CdcStartPosition=str(oracle_scn),
                )["ReplicationTask"]

                dms.start_replication_task(
                    ReplicationTaskArn=new_task["ReplicationTaskArn"],
                    StartReplicationTaskType="start-replication",
                )

                message = (
                    f"Created recovery task {new_task['ReplicationTaskArn']} "
                    f"from SCN {oracle_scn}"
                )

        conn.commit()
        cur.close()
        conn.close()

        return {
            "statusCode": 200,
            "body": json.dumps(
                {
                    "task_arn": task_arn,
                    "status": status,
                    "source_latency": source_latency,
                    "target_latency": target_latency,
                    "message": message,
                }
            ),
        }

    except Exception as exc:
        return {"statusCode": 500, "body": json.dumps({"error": str(exc)})}

Supporting tables:

CREATE TABLE IF NOT EXISTS admin.dms_monitoring (
  monitoring_id   SERIAL PRIMARY KEY,
  monitoring_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  task_arn        TEXT,
  status          TEXT,
  source_latency  NUMERIC,
  target_latency  NUMERIC
);

CREATE TABLE IF NOT EXISTS admin.dms_recovery_events (
  event_id        SERIAL PRIMARY KEY,
  event_time      TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  task_arn        TEXT,
  event_type      TEXT,
  checkpoint_name TEXT,
  oracle_scn      NUMERIC,
  notes           TEXT
);

Scheduled with an EventBridge rule:

aws events put-rule \
  --name "DMSCDCMonitoringRule" \
  --schedule-expression "rate(5 minutes)" \
  --state ENABLED

aws events put-targets \
  --rule "DMSCDCMonitoringRule" \
  --targets "Id"="1","Arn"="arn:aws:lambda:region:account:function:CDCMonitor"

aws lambda add-permission \
  --function-name CDCMonitor \
  --statement-id EventBridgeSchedule \
  --action 'lambda:InvokeFunction' \
  --principal events.amazonaws.com \
  --source-arn arn:aws:events:region:account:rule/DMSCDCMonitoringRule

Part 5: Long-term maintenance

Oracle: archive log space

The recovery window closes silently when the fast recovery area fills, so monitor it and alert before it does.

CREATE TABLE admin.archive_log_monitor (
  monitor_id      NUMBER GENERATED ALWAYS AS IDENTITY,
  monitor_time    TIMESTAMP DEFAULT SYSTIMESTAMP,
  total_space_gb  NUMBER,
  used_space_gb   NUMBER,
  percent_used    NUMBER,
  oldest_log_date DATE,
  PRIMARY KEY (monitor_id)
);

CREATE TABLE admin.dms_alerts (
  alert_id     NUMBER GENERATED ALWAYS AS IDENTITY,
  alert_time   TIMESTAMP DEFAULT SYSTIMESTAMP,
  alert_type   VARCHAR2(50),
  severity     VARCHAR2(20),
  message      VARCHAR2(4000),
  acknowledged CHAR(1) DEFAULT 'N',
  PRIMARY KEY (alert_id)
);

CREATE OR REPLACE PROCEDURE admin.monitor_archive_logs
AS
  v_total_space     NUMBER;
  v_used_space      NUMBER;
  v_percent_used    NUMBER;
  v_oldest_log_date DATE;
  v_alert_threshold NUMBER := 80;
BEGIN
  SELECT a.space_limit / 1024 / 1024 / 1024,
         a.space_used  / 1024 / 1024 / 1024,
         ROUND(a.space_used / a.space_limit * 100, 2)
  INTO   v_total_space, v_used_space, v_percent_used
  FROM   v$recovery_file_dest a;

  SELECT MIN(first_time) INTO v_oldest_log_date
  FROM   v$archived_log
  WHERE  archived = 'YES' AND deleted = 'NO';

  INSERT INTO admin.archive_log_monitor
    (total_space_gb, used_space_gb, percent_used, oldest_log_date)
  VALUES (v_total_space, v_used_space, v_percent_used, v_oldest_log_date);

  IF v_percent_used > v_alert_threshold THEN
    INSERT INTO admin.dms_alerts (alert_type, severity, message)
    VALUES ('ARCHIVE_LOG_SPACE', 'WARNING',
            'Archive log space at ' || v_percent_used || '% capacity. Oldest log: '
            || TO_CHAR(v_oldest_log_date, 'YYYY-MM-DD HH24:MI:SS'));
  END IF;

  COMMIT;
EXCEPTION
  WHEN OTHERS THEN
    INSERT INTO admin.dms_alerts (alert_type, severity, message)
    VALUES ('ARCHIVE_LOG_SPACE', 'ERROR',
            'Error monitoring archive logs: ' || SQLERRM);
    COMMIT;
END;
/

BEGIN
  DBMS_SCHEDULER.CREATE_JOB (
    job_name        => 'ARCHIVE_LOG_MONITOR_JOB',
    job_type        => 'STORED_PROCEDURE',
    job_action      => 'admin.monitor_archive_logs',
    start_date      => SYSTIMESTAMP,
    repeat_interval => 'FREQ=HOURLY;INTERVAL=1',
    enabled         => TRUE,
    comments        => 'Monitor archive log space hourly',
    auto_drop       => FALSE
  );
END;
/

Writing a row into admin.dms_alerts is not an alert. Point something at that table — or, better, publish the same condition as a CloudWatch metric — so that the threshold being crossed reaches a person.

PostgreSQL: archiving the monitoring tables

The monitoring table grows continuously, so move each completed month into its own table and drop old checkpoints.

CREATE TABLE IF NOT EXISTS admin.maintenance_log (
  log_id           SERIAL PRIMARY KEY,
  log_time         TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  maintenance_type TEXT,
  details          TEXT
);

CREATE OR REPLACE PROCEDURE admin.perform_dms_maintenance()
LANGUAGE plpgsql
AS $$
DECLARE
  v_old_month TEXT := to_char(current_date - interval '1 month', 'YYYY_MM');
  v_old_start DATE := date_trunc('month', current_date - interval '1 month')::date;
  v_old_end   DATE := date_trunc('month', current_date)::date;
  v_archive   TEXT;
  v_moved     BIGINT;
BEGIN
  v_archive := 'dms_monitoring_' || v_old_month;

  EXECUTE format(
    'CREATE TABLE IF NOT EXISTS admin.%I (LIKE admin.dms_monitoring INCLUDING ALL)',
    v_archive);

  EXECUTE format(
    'INSERT INTO admin.%I SELECT * FROM admin.dms_monitoring
      WHERE monitoring_time >= %L AND monitoring_time < %L
      ON CONFLICT DO NOTHING',
    v_archive, v_old_start, v_old_end);
  GET DIAGNOSTICS v_moved = ROW_COUNT;

  EXECUTE format(
    'DELETE FROM admin.dms_monitoring
      WHERE monitoring_time >= %L AND monitoring_time < %L',
    v_old_start, v_old_end);

  DELETE FROM admin.dms_cdc_checkpoints
  WHERE checkpoint_time < current_date - interval '90 days';

  INSERT INTO admin.maintenance_log (maintenance_type, details)
  VALUES ('DMS_TABLES',
          format('Archived %s rows to admin.%s', v_moved, v_archive));
END;
$$;

The month is derived, not written in. A version with a literal table name and a literal month — INSERT INTO admin.dms_monitoring_2025_03 … WHERE date_trunc('month', monitoring_time) = DATE '2025-03-01' — does nothing useful once that month has passed, and then deletes rows from a table it has just created for a different month.

VACUUM is deliberately absent: it cannot run inside a transaction block, and PL/pgSQL always executes inside one. Run it from the caller instead:

psql -v ON_ERROR_STOP=1 -c "CALL admin.perform_dms_maintenance();"
psql -v ON_ERROR_STOP=1 -c "VACUUM ANALYZE admin.dms_monitoring;"
psql -v ON_ERROR_STOP=1 -c "VACUUM ANALYZE admin.dms_cdc_checkpoints;"
psql -v ON_ERROR_STOP=1 -c "VACUUM ANALYZE admin.dms_recovery_events;"

There is no exception handler in the procedure. One that logged the failure and re-raised would lose its own log row to the rollback; leaving the error to propagate means the scheduler that ran it reports the failure, which is where it can be seen.

Health checks

An Oracle-side check, run on the database host under OS authentication so that no password is handled at all:

#!/bin/bash
set -euo pipefail

export ORACLE_SID="your_oracle_sid"
export ORACLE_HOME="/path/to/oracle/home"

echo "======== Oracle CDC health check: $(date) ========"

"$ORACLE_HOME/bin/sqlplus" -s / as sysdba <<'EOF'
SET LINESIZE 200 PAGESIZE 100 FEEDBACK OFF VERIFY OFF

PROMPT -- Archive log mode and supplemental logging
SELECT log_mode, supplemental_log_data_min, supplemental_log_data_pk FROM v$database;

PROMPT -- Fast recovery area usage
SELECT name,
       ROUND(space_limit / 1024 / 1024 / 1024, 2) AS limit_gb,
       ROUND(space_used  / 1024 / 1024 / 1024, 2) AS used_gb,
       ROUND(space_reclaimable / 1024 / 1024 / 1024, 2) AS reclaimable_gb,
       ROUND(space_used / space_limit * 100, 2) AS percent_used
FROM v$recovery_file_dest;

PROMPT -- Archive log generation, last 24 hours
SELECT TRUNC(completion_time, 'HH') AS hour,
       COUNT(*) AS logs_generated,
       ROUND(SUM(blocks * block_size) / 1024 / 1024, 2) AS size_mb
FROM v$archived_log
WHERE completion_time > SYSDATE - 1
GROUP BY TRUNC(completion_time, 'HH')
ORDER BY hour;

PROMPT -- SCN checkpoints recorded in the last 24 hours
SELECT checkpoint_name,
       TO_CHAR(checkpoint_time, 'YYYY-MM-DD HH24:MI:SS') AS captured_at,
       scn
FROM admin.dms_scn_checkpoints
WHERE checkpoint_time > SYSDATE - 1
ORDER BY checkpoint_time;
EXIT;
EOF

And a target-side check. The password comes from a ~/.pgpass entry owned by the account running the script, so it appears neither in the file nor in the process list:

#!/bin/bash
set -euo pipefail

PG_HOST="postgres-host"
PG_PORT="5432"
PG_DB="postgres-db"
PG_USER="dms_monitor"

echo "======== PostgreSQL CDC health check: $(date) ========"

aws dms describe-replication-tasks \
  --query 'ReplicationTasks[*].{Task:ReplicationTaskIdentifier,Status:Status,Type:MigrationType,SourceLatency:ReplicationTaskStats.CdcLatencySource,TargetLatency:ReplicationTaskStats.CdcLatencyTarget}' \
  --output table

psql -h "$PG_HOST" -p "$PG_PORT" -U "$PG_USER" -d "$PG_DB" -v ON_ERROR_STOP=1 <<'EOF'
\echo '-- Recent CDC checkpoints'
SELECT checkpoint_id, checkpoint_time, checkpoint_name, oracle_scn
FROM admin.dms_cdc_checkpoints ORDER BY checkpoint_time DESC LIMIT 5;

\echo '-- Recent recovery events'
SELECT event_id, event_time, event_type, checkpoint_name, oracle_scn
FROM admin.dms_recovery_events ORDER BY event_time DESC LIMIT 5;

\echo '-- CDC latency, last 24 hours'
SELECT date_trunc('hour', monitoring_time) AS hour,
       ROUND(AVG(source_latency), 1) AS avg_source,
       MAX(source_latency)           AS max_source,
       ROUND(AVG(target_latency), 1) AS avg_target,
       MAX(target_latency)           AS max_target
FROM admin.dms_monitoring
WHERE monitoring_time > CURRENT_TIMESTAMP - INTERVAL '24 hours'
GROUP BY 1 ORDER BY 1;
EOF

Backing up what recovery depends on

Archived redo logs are the recovery window. Back them up before they are reclaimed, and only delete what has been backed up:

#!/bin/bash
set -euo pipefail

export ORACLE_SID="your_oracle_sid"
export ORACLE_HOME="/path/to/oracle/home"
BACKUP_DIR="/backup/archivelogs"
DAYS_TO_KEEP=7

BACKUP_FOLDER="$BACKUP_DIR/$(date +%Y%m%d)"
mkdir -p "$BACKUP_FOLDER"

"$ORACLE_HOME/bin/rman" <<EOF
CONNECT TARGET /
BACKUP ARCHIVELOG ALL NOT BACKED UP FORMAT '$BACKUP_FOLDER/%d_%T_%s_%p.arc';
DELETE NOPROMPT ARCHIVELOG ALL COMPLETED BEFORE 'SYSDATE-$DAYS_TO_KEEP'
  BACKED UP 2 TIMES TO DEVICE TYPE DISK;
EXIT;
EOF

find "$BACKUP_DIR" -mindepth 1 -maxdepth 1 -type d -mtime +$DAYS_TO_KEEP -exec rm -rf {} +

The DMS configuration itself is worth exporting too, so that a replacement task can be built without reconstructing its settings from memory:

#!/bin/bash
set -euo pipefail

BACKUP_DIR="/backup/dms_config"
AWS_REGION="your-region"
BACKUP_TIME=$(date +%Y%m%d_%H%M%S)
BACKUP_FOLDER="$BACKUP_DIR/$BACKUP_TIME"
mkdir -p "$BACKUP_FOLDER"

aws dms describe-endpoints --region "$AWS_REGION" \
  > "$BACKUP_FOLDER/dms_endpoints.json"
aws dms describe-replication-instances --region "$AWS_REGION" \
  > "$BACKUP_FOLDER/dms_replication_instances.json"
aws dms describe-replication-tasks --region "$AWS_REGION" \
  > "$BACKUP_FOLDER/dms_replication_tasks.json"

psql -h postgres-host -p 5432 -U dms_monitor -d postgres-db \
  -c "\COPY (SELECT * FROM admin.dms_cdc_checkpoints ORDER BY checkpoint_time)
      TO '$BACKUP_FOLDER/dms_cdc_checkpoints.csv' WITH CSV HEADER"

tar -czf "$BACKUP_DIR/dms_backup_$BACKUP_TIME.tar.gz" -C "$BACKUP_DIR" "$BACKUP_TIME"
rm -rf "$BACKUP_FOLDER"

find "$BACKUP_DIR" -name "dms_backup_*.tar.gz" -type f -mtime +30 -delete

describe-endpoints returns endpoint configuration, not credentials — the passwords stay in Secrets Manager, which is where the endpoints should be reading them from in the first place.

Part 6: Troubleshooting

ORA-01291: missing logfile

DMS is asking for an archived log the source no longer has.

SELECT sequence#, name, status
FROM v$archived_log
WHERE sequence# = <sequence_number_from_error>;

If the log was backed up, restore it and restart the task. If it was not, the position is unrecoverable; see Part 2, Step 4.

CDC latency rising

Establish first whether the source is generating redo faster than DMS is consuming it:

-- On the source, under OS authentication: sqlplus -s / as sysdba
SELECT TRUNC(completion_time, 'HH') AS hour,
       COUNT(*) AS logs_generated,
       ROUND(SUM(blocks * block_size) / 1024 / 1024, 2) AS size_mb
FROM v$archived_log
WHERE completion_time > SYSDATE - 1
GROUP BY TRUNC(completion_time, 'HH')
ORDER BY hour;

aws cloudwatch get-metric-statistics \
  --namespace AWS/DMS \
  --metric-name CDCLatencySource \
  --dimensions Name=ReplicationInstanceIdentifier,Value=your-instance-id \
               Name=ReplicationTaskIdentifier,Value=your-task-id \
  --start-time "$(date -u -d '24 hours ago' +%Y-%m-%dT%H:%M:%SZ)" \
  --end-time "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
  --period 3600 \
  --statistics Average

If DMS is the bottleneck, scale the replication instance:

aws dms modify-replication-instance \
  --replication-instance-arn <instance-arn> \
  --replication-instance-class dms.c5.xlarge

Binary Reader instead of LogMiner is the other lever where redo volume is the problem — see GoldenGate to PostgreSQL with DMS.

SCN jumping unexpectedly

SELECT a.checkpoint_id,
       a.checkpoint_time,
       a.scn,
       a.scn - LAG(a.scn) OVER (ORDER BY a.checkpoint_time) AS scn_jump
FROM admin.dms_scn_checkpoints a
WHERE a.checkpoint_time > SYSDATE - 7
ORDER BY a.checkpoint_time;

A large jump usually corresponds to a bulk operation. Cross-check against dba_scheduler_job_run_details for jobs running at that time.

Inconsistent data after recovery

Row counts agreeing does not mean the data is right. Check a business invariant over the rows touched since the recovery point:

WITH order_totals AS (
  SELECT o.order_id,
         o.total_amount,
         SUM(oi.price * oi.quantity) AS calculated_total
  FROM orders o
  JOIN order_items oi ON o.order_id = oi.order_id
  WHERE o.updated_at > TIMESTAMP '2025-02-15 14:30:00'
  GROUP BY o.order_id, o.total_amount
)
SELECT order_id, total_amount, calculated_total,
       total_amount - calculated_total AS difference
FROM order_totals
WHERE ABS(total_amount - calculated_total) > 0.01;

Duplicated rows are the most common finding, and they follow directly from DMS’s at-least-once delivery: a table replicated without a primary key or unique index has no way to recognise a change it has already applied.