feat(jobs): add dataset_cleanup_sweep global job

[SgtPooki]

What changed

New scheduled global job dataset_cleanup_sweep that periodically reconciles Deal.cleaned_up with FWSS state. For every uncleaned Deal row with a non-null data_set_id, the sweeper probes FWSS getDataSet(dataSetId):

• pdpEndEpoch != 0n → flip cleaned_up=true (terminated)
• getDataSet returns null → flip cleaned_up=true (dataset removed)
• otherwise → leave alone (live)
• probe error → skip + log + retry next tick

UPDATEs filter cleaned_up=false and run inside a single transaction per bucket. Reuses the existing UPDATE block from repairTerminatedDataSet via a new shared method DealService.markDealsCleanedUpForDataSets.

Why

In session-key + multisig payer mode, dealbot cannot auto-terminate datasets (#546). Operators run terminateService via Safe (precedent: #545). After the Safe batch lands, synapse-sdk's createContext filters out the terminated dataset, so getDataSetProvisioningStatus returns "missing" instead of "terminated" and repairTerminatedDataSet is never invoked for those rows. The retrieval candidate selector keeps picking the stale Deal rows and pollutes failure metrics.

Empirically on calibration staging (7-day window):

• 4530 successful retrievals, 398 failed
• 340 / 398 failed retrievals (87%) tied to FWSS-terminated datasets
• 6 / 398 tied to FWSS-DNE datasets
• Two datasets account for 337/340 of the terminated pollution

How to verify

pnpm test from repo root (364 tests pass).

Manual: in staging after deploy, watch for dataset_cleanup_sweep_completed log events. First run drains the historical backlog; steady-state ticks should report datasetsTerminated: 0 after the queue is clean.

Notes / risks

• Draft. New unit tests for the sweeper + handler are a follow-up.
• Cadence is 24h by default (DATASET_CLEANUP_SWEEP_INTERVAL_SECONDS=86400). Operators can dial down via env if a backlog spikes after a Safe batch.
• Partial index migration: CREATE INDEX idx_deals_unclean_dataset ON deals(data_set_id) WHERE cleaned_up = false AND data_set_id IS NOT NULL. Without it, the SELECT DISTINCT data_set_id WHERE cleaned_up = false triggers a full table scan on every tick.
• pdpEndEpoch != 0n is treated as irreversible (verified against FWSS source: no path resets it to zero once set).
• Multicall errors are bucketed separately from result == null so a flaky RPC will not wrongly mark deals cleaned up.
• Out of scope for this PR: Deal rows with data_set_id IS NULL (separate root cause from Synapse upgrade causes Dealbot to create a new dataset per deal for providers beyond first 100 client datasets #511 piece_id=0 pattern). Filed as a separate concern.
• Follow-up optimization noted in Dealbot can't auto-terminate datasets when payer is a Safe multisig #546 review: subgraph already exposes pdpEndEpoch per dataset, so a future iteration could skip the per-dataset multicall entirely.

Refs

• Dealbot can't auto-terminate datasets when payer is a Safe multisig #546 root cause tracker (safeReadTransport + multisig payer interaction)
• Terminate 197 stuck provider-24 datasets on calibnet via Safe multisig #545 manual Safe cleanup of 197 stuck provider-24 datasets (workflow this sweeper closes)
• PR fix(jobs): defer terminate to operator when payer is multisig #547 short-term DataSetTerminateRequiresOperatorError (related, not a dependency)
• PR fix: work around Lotus eth_call rejection for multisig accounts #419 safeReadTransport workaround (background)
• fix(eth): allow eth_call and eth_estimateGas from contract and non-existent addresses filecoin-project/lotus#13470 (background)