Staged uploads live in a process-local dict on the store #1489

Problem

PgReportingLedgerStore.stage_revision_rows stores the StagedRevisionRows handle in self._staged_rows, keyed by (account_id, revision_id, content_sha256), and commit_revision pops it later. Every failure path between the two calls must call discard_staged_revision_rows, or the entry (which holds the prepared chunks) lives for the rest of the process. The producer is the only caller, and it already has the handle in scope, so the dict adds state without adding information. The store protocol does not declare the staging API, which is why the producer reaches it through getattr and a type: ignore.

Evidence

All references are to main at 47084e83a.

• src/adcp/reporting/ledger/pg.py:251:

  self._staged_rows: dict[tuple[str, str, str], Any] = {}

• pg.py:1383: self._staged_rows[_staged_key(revision)] = staged at the end of stage_revision_rows.

• pg.py:1287: self._staged_rows.pop(_staged_key(revision), None) inside commit_revision.

• pg.py:1386-1388: discard_staged_revision_rows exists only to pop the entry: "Forget staged chunks of a revision that will not commit; the intent expires."

• src/adcp/reporting/ledger/producer.py:1866-1874: the producer calls stage_revision_rows through getattr(self._store, "stage_revision_rows", None), and in a finally block calls self._store.discard_staged_revision_rows(revision) # type: ignore[attr-defined].

Suggested fix

Return the handle from stage_revision_rows and accept it as an argument of commit_revision (for example commit_revision(revision, rows, staged=...)). The producer passes what it received; nothing is stored on the store, and discard_staged_revision_rows and _staged_rows go away. Declare both methods on the store protocol so the getattr and the type: ignore in the producer can be removed.

Introduced in #1467.