perf: Production readiness with Workload purging, and AuthN injection

[mdrakiburrahman]

Why this change is needed

Closes #33.

The mon_database_metadata incremental model in dbt-ingest-geneva failed
during a long stress run with:

Compilation Error in model mon_database_metadata (models/mon_database_metadata.sql)
  Object of type datetime is not JSON serializable

Inside the adapter:

scope adapter: write_batch_sources failed for batch 20

The crash always happens on the second parquet compaction
(batch_id > 0 and batch_id % source_compaction_interval == 0),
i.e. once a previous .parquet snapshot already exists under
_checkpoint/sources/. With the default source_compaction_interval=10,
that's batch 20 (as observed); with source_compaction_interval=1 (the
aggressive case), it's batch 2.

Root cause

CheckpointManager._write_snapshot_parquet builds the next snapshot by:

1. Reading the latest existing .parquet snapshot via DuckDB
   (SELECT * FROM read_parquet(...)).
2. Reading JSONL diffs written since that snapshot.
3. Materialising everything into a list[dict] in Python.
4. Writing the merged list as NDJSON via
   nf.write(json.dumps(r) + "\n"), then loading the NDJSON back via
   read_json_auto and COPY ... TO ... (FORMAT PARQUET).

batchProcessingTime is created as an ISO 8601 string in
_build_source_records, but DuckDB's read_json_auto infers it as
TIMESTAMP once the NDJSON has enough rows of consistent ISO text
(reproducible at ~20 rows with duckdb 1.5.3; the in-test fast-loop case
with ≤6 records per batch stays under the threshold, which is why our
existing multi-compaction tests didn't catch it).

So the first snapshot (batch 10) stores the column as TIMESTAMP. On the
next compaction (batch 20), DuckDB returns those values as Python
datetime.datetime, and json.dumps(r) then raises
TypeError: Object of type datetime is not JSON serializable. The outer
try/except in write_batch_sources logs the failure and re-raises,
which dbt surfaces as a "Compilation Error" on the model after ~7h43m of
wasted SCOPE compute.

How

Two small changes in dbt/adapters/scope/checkpoint.py:

1. Module-level _json_default(o) helper that maps
   datetime → o.isoformat() and re-raises TypeError for anything else.
   Naive datetime instances (which is what DuckDB hands back after a
   TIMESTAMP round-trip) are re-attached to UTC before formatting so
   the on-disk ISO string keeps its +00:00 suffix — preserving the
   timezone-aware contract that _build_source_records originally
   established. Wired into both json.dumps call sites — _write_jsonl
   (defensive; today's inputs are pure str/int) and the NDJSON write in
   _write_snapshot_parquet (where the crash actually happens). This
   covers any existing customer parquet snapshots whose
   batchProcessingTime column is already TIMESTAMP.
2. Stabilise the snapshot parquet schema by replacing the snapshot's
   SELECT * with an explicit, fully-typed projection — path VARCHAR,
   modificationTime BIGINT, batchId INTEGER,
   batchProcessingTime TIMESTAMP. This stops the schema from drifting
   between VARCHAR and TIMESTAMP depending on DuckDB's sample-size
   heuristic, and also defends against future drift on the other three
   columns. Every subsequent read then returns datetime for
   batchProcessingTime, and _json_default normalises it back to an
   ISO string on the next round-trip.

Considerations

• No data migration required. Existing parquet snapshots already on
  customer ADLS accounts (where the column was previously inferred as
  TIMESTAMP) are now handled correctly by _json_default.
• Tiny diff, lowest blast radius. A larger refactor (doing the
  union entirely in DuckDB via read_parquet UNION ALL read_json_auto
  to skip the Python round-trip) would be cleaner but is out of scope —
  the compaction code path is I/O-bound on ADLS, not on the Python
  round-trip, so the perf win at current scale (≤a few hundred records
  per snapshot) is negligible.
• Alternative considered: instead of pinning to TIMESTAMP, we
  could pin to VARCHAR — both fix the bug. We picked TIMESTAMP
  because it's the better long-term schema (and matches what DuckDB
  picks at production scale anyway).

Test

• New regression unit test:
  tests/unit/test_checkpoint_lifecycle.py::TestCheckpointLifecycle::test_compaction_handles_timestamp_typed_prior_snapshot
  pre-seeds a TIMESTAMP-typed parquet snapshot (22 rows) into the
  in-memory ADLS mock, writes an intermediate JSONL diff at batch 15
  (3 rows), then triggers compaction at batch 20 (2 rows) and asserts:
  • write_batch_sources does not raise.
  • The new snapshot contains records from all three legs of the
    union — prior snapshot + JSONL diff + current batch
    (22 + 3 + 2 = 27 rows, with batchId values for every leg).
  • The new snapshot's batchProcessingTime is deterministically
    TIMESTAMP.
    Verified to fail without the fix with the exact production error
    (TypeError: Object of type datetime is not JSON serializable) and
    pass with it.
• .scripts/run.sh unit-test — 336/336 passing.
• .scripts/run.sh lint — clean.
• .scripts/run.sh integration-test — full suite re-run end-to-end
  against ADLA (no regressions observed).

---

addendum: credential retry resilience (commit 9fc39e8)

Why a second fix in this PR

After the datetime fix landed, the same dbt-ingest-geneva stress run
surfaced a second, independent failure mode in a CI workflow run:
https://github.com/microsoft/dbt-scope/actions/runs/27011393672/job/79716601894

        except Exception as ex:
            # could be a timeout, for example
            error = CredentialUnavailableError(message="Failed to invoke the Azure CLI")
>           raise error from ex
E           azure.identity._exceptions.CredentialUnavailableError: Failed to invoke the Azure CLI

Under high concurrency (the production model runs with
Concurrency: 32 threads), the az subprocess that
AzureCliCredential.get_token shells out to can time out. The adapter
previously raised on the first occurrence, failing the dbt run after
multi-hour ingestion jobs even though the underlying issue is
transient.

What changed

Centralised retry handling in LockedTokenCredential:

• New RetryPolicy dataclass — linear backoff, capped delay, no
  jitter. Total attempts == max_retries + 1.
• RetryPolicy.from_http_retries(http_retries) reuses the existing
  http_retries profile field so users can tune retries via
  profiles.yml (default bumped 3 → 10 per user spec).
• LockedTokenCredential.get_token catches only
  CredentialUnavailableError (transient subprocess failures);
  permanent ClientAuthenticationError is intentionally not retried.
• Releases the FileLock between attempts so other workers continue
  to make progress.
• Sleeps via an injectable callable for deterministic tests.

All credential acquisition sites now flow through
LockedTokenCredential and honour the policy:

• ScopeConnectionHandle (ADLA query path)
• AdlsGen1Client (source-file discovery)
• CheckpointManager (watermark + snapshot read/write)
• ScopeAdapter.list_relations_without_caching

Defensive: no more silent swallowing of exhausted credentials

A rubber-duck review surfaced several broad except Exception blocks
that would silently convert exhausted credentials into apparently
benign outcomes — masking auth failures and corrupting incremental
state. Each now re-raises CredentialUnavailableError before the
broad handler:

• CheckpointManager.read_watermark → would have returned None,
  silently flipping incremental → full refresh.
• AdlsGen1Client._list_one_dir / _walk / list_files (non-recursive)
  / _directory_exists / _list_directory_files / enrich_with_estimates
  → would have returned empty / partial file lists, advancing the
  watermark past unseen files.
• DuckDbDeltaLakeClient.table_exists / get_max_partition /
  get_columns / list_table_paths → would have looked like
  "Delta table missing".
• ScopeAdapter.list_relations_without_caching (inner + outer except)
  → would have returned an empty Delta-table list.

Tests

• New unit test file tests/unit/test_locked_token_credential.py
  (14 tests): defaults, from_http_retries for None/-1/0/25,
  succeed-on-first-attempt (no sleep), succeed-after-N-failures,
  exhaust-and-re-raise, linear-capped delay sequence
  [1,2,3,...,10,10,10], zero-retries == single-attempt,
  non-CredentialUnavailableError exceptions are NOT retried,
  claims= kwarg passthrough, lock-released-between-attempts (via
  parallel acquisition probe), default policy used when None.
• New regression tests:
  • tests/unit/test_adls_gen1_client.py::TestAdlsGen1Client::test_recursive_subdir_credential_exhaustion_propagates
  • tests/unit/test_adls_gen1_client.py::TestAdlsGen1Client::test_non_recursive_credential_exhaustion_propagates
  • tests/unit/test_checkpoint.py::TestCheckpointManagerWatermark::test_read_watermark_propagates_credential_exhaustion
• .scripts/run.sh unit-test — 353/353 passing.
• .scripts/run.sh lint — clean.
• .scripts/run.sh integration-test — 3/3 passing end-to-end
  against ADLA (9:58).

[mdrakiburrahman]

Pushed second commit 9fc39e8 adding credential-retry resilience for the CredentialUnavailableError: Failed to invoke the Azure CLI failure mode seen in CI run 27011393672. See updated PR body for details.

Verification: 353/353 unit tests, 3/3 integration tests (9:58), lint clean.