test: add target-neutral CDC data-integrity oracle

[viggy28]

Summary

Add an end-to-end CDC data-integrity oracle that compares source Postgres with Streambed's durable target while bypassing Streambed's pgwire query layer.

The same deterministic scenario runs against Iceberg and DuckLake. Destination data is read directly with embedded DuckDB using iceberg_scan(...) or a read-only DuckLake catalog attachment.

Closes #86.

Implementation

• Build and start the real streambed sync process for each target.
• Use isolated replication slots, state stores, S3 prefixes, and DuckLake catalogs.
• Capture a Postgres WAL barrier after every source transaction and wait for confirmed_flush_lsn instead of sleeping.
• Verify each table's target last_flush_lsn at every checkpoint.
• Compare complete rows keyed by primary or composite key.
• Detect missing rows, extra rows, duplicate keys, missing columns, null mismatches, and per-column value mismatches.
• Normalize values by source logical type:
  • exact booleans, integers, text, and binary values;
  • exact rational decimal comparison without float64;
  • timestamp and timestamptz semantics handled separately;
  • canonical UUIDs;
  • structural JSON normalization with lossless JSON numbers.
• Explicitly disable any inherited query-server address so the test cannot accidentally exercise pgwire.
• Open DuckLake readers in read-only mode.

Scenario coverage

Three durable checkpoints cover:

1. Initial inserts, composite keys, null-heavy rows, large JSON integers, timestamp types, UUIDs, and binary edge cases.
2. Regular updates, primary-key changes, several updates to one key before flush, delete/reinsert, and a transaction touching multiple tables.
3. Streambed restart/resume followed by updates, deletes, and inserts.

Forced post-commit/pre-ack replay remains covered by the existing failpoint integration tests.

Validation

Started fresh Postgres and MinIO containers, created the bucket, built Streambed, and ran the oracle through both targets:

docker compose -f test/integration/docker-compose.yml down -v
docker compose -f test/integration/docker-compose.yml up -d postgres minio --wait
docker compose -f test/integration/docker-compose.yml up createbucket
go test -tags integration -v -count=1 -timeout 8m ./test/integration \
  -run '^TestCDC(DataIntegrityOracle|OracleNormalization|OracleRejectsDuplicateKeys)$'

Result: PASS for all three checkpoints on Iceberg and DuckLake.

Also ran the complete PR integration-smoke command successfully:

go test -tags integration -short -v -count=1 -timeout 8m \
  -run 'TestEndToEnd$|TestEndToEndUpdateDelete$|TestSchemaEvolution_AddColumn$|TestMutationModesDuckDBCorrectness/cow$|TestCDCDataIntegrityOracle$' \
  ./test/integration

And:

go test -race ./internal/... ./config/...
go test -tags integration ./test/integration -run '^$'
git diff --check

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 6b5976f7ca

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Run the CDC oracle outside the slash-split regexp

In the inspected integration-smoke command, the new oracle is never selected. go help testflag states that -run expressions are split at each unbracketed /; consequently, TestCDCDataIntegrityOracle$ becomes part of the second, subtest-name expression, while the top-level test must match the first expression ending at TestMutationModesDuckDBCorrectness. Run the oracle separately or escape/bracket the slash so the complete expression is not split.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Reject trailing data after the JSON value

When a target JSON string contains the expected value followed by additional data, such as {"a":1}null, this single Decode succeeds on the first value and the canonical result matches the source, allowing the data-integrity oracle to pass corrupted target data. After decoding the value, perform another decode and require io.EOF so the entire target string is validated.

Useful? React with 👍 / 👎.