GCS Payload Reconciliation Pipeline

Summary

When syncserver offloads a large BSO payload to Google Cloud Storage (syncserver/src/web/payload_offload.rs), the GCS object is created with custom metadata committed=false and customTime=now, and the BSO row’s payload_link column points at it. A separate pipeline must:

• Finalize newly-committed objects — flip metadata to committed=true and pin customTime to its maximum value (2200-12-31T23:59:59Z) so the bucket’s lifecycle policy (see below) cannot reclaim them.
• Garbage-collect orphans — delete GCS objects whose row’s payload_link was replaced (UPDATE) or removed (DELETE, including Spanner row-deletion-policy TTL deletes).

An object whose write never reached a Spanner commit never receives a finalize, and there are two shapes of that. If the database transaction fails or rolls back, syncserver issues an inline best-effort delete of the objects it uploaded for that request and ignores the result. If an upload itself fails, the request is abandoned fail-fast and objects already uploaded for it are not cleaned up inline at all. Either way an object can be left stranded at committed=false with its upload-time customTime.

Anything that slips through is reaped by a GCS lifecycle policy (bucket.tf in webservices-infra/sync) that deletes objects whose customTime is more than 30 days old. Flipping customTime to the max sentinel is what protects committed objects from that policy: daysSinceCustomTime goes permanently negative once finalized, so the policy cannot touch them regardless of object age.

The 30 day window is the safety margin for the whole asynchronous arm. It has to comfortably exceed the worst case time from upload to finalize, which is set by the cronjob cadence plus any Pub/Sub or Dataflow backlog. At a five minute cadence the margin is four orders of magnitude, so the number only becomes interesting if the pipeline is stopped for weeks.

This document covers that pipeline. It consumes the payload_link_changes Spanner change stream defined in syncstorage-spanner/src/schema.ddl.

Architecture

flowchart LR
    stream[("Spanner change stream
    payload_link_changes
    OLD_AND_NEW_VALUES, 7d retention")]

    subgraph dataflow["Custom Dataflow flex template"]
        filter{"payload_link NULL
        on both sides?"}
    end

    drop(["dropped"])
    topic["Pub/Sub
    payload-link-changes"]
    dlq["Pub/Sub
    payload-link-changes-dlq"]

    subgraph reconciler["Reconciler cronjob"]
        route{"which side
        carries a link?"}
        finalize["finalize the object
        committed=true
        customTime=MAX"]
        delete["delete the object"]
    end

    stream -->|DataChangeRecord| filter
    filter -->|yes| drop
    filter -->|"no, publish as JSON"| topic
    topic -->|pull subscription| route
    topic -.->|"after 5 failed deliveries"| dlq
    route -->|new link| finalize
    route -->|"old link, replaced or removed"| delete

Both reconciler actions are idempotent, so a redelivered record is harmless.

Components

1. Spanner change stream — payload_link_changes

Defined in syncstorage-spanner/src/schema.ddl:

CREATE CHANGE STREAM payload_link_changes
    FOR bsos(payload_link), batch_bsos(payload_link)
    OPTIONS (
      retention_period = '7d',
      value_capture_type = 'OLD_AND_NEW_VALUES'
    );

Column-scoped: an UPDATE that does not touch payload_link produces no record. INSERTs and DELETEs always produce a record, even when payload_link is NULL — those are dropped at the next stage.

The Spanner DDL is not auto-applied; run gcloud spanner databases ddl update against the target database after merging.

2. Custom Dataflow flex template — tools/payload-link-dataflow/

A standalone Apache Beam pipeline (Java, Beam 2.60.0) that:

1. Reads payload_link_changes via SpannerIO.readChangeStream().
2. Applies a Filter.by(isPayloadLinkActionable) step that drops records whose every mod has payload_link NULL on both sides. Malformed records pass through so the reconciler / DLQ surfaces them — not the filter.
3. Serializes each surviving DataChangeRecord to JSON and publishes to a Pub/Sub topic.

The pipeline is not vendored from GoogleCloudPlatform/DataflowTemplates. We own a small standalone source tree under src/; the upstream Cloud_Spanner_Change_Streams_to_PubSub template is referenced for intent comparison via upstream-customization.patch (documentation only — not a build input).

Build / publish (operator runs from webservices-infra):

docker build -t <REGISTRY>/syncserver-payload-link-dataflow:<TAG> \
  tools/payload-link-dataflow
docker push <REGISTRY>/syncserver-payload-link-dataflow:<TAG>

gcloud dataflow flex-template build \
  gs://<BUCKET>/templates/syncserver-payload-link-dataflow.json \
  --image <REGISTRY>/syncserver-payload-link-dataflow:<TAG> \
  --sdk-language JAVA \
  --metadata-file tools/payload-link-dataflow/metadata.json

Launch parameters (full list in tools/payload-link-dataflow/metadata.json):

• spannerProjectId, spannerInstanceId, spannerDatabase — the syncstorage Spanner database.
• spannerMetadataInstanceId, spannerMetadataDatabase — where the change-stream connector keeps its partition-state table. Recommend a dedicated database in prod for isolation.
• changeStreamName=payload_link_changes.
• spannerDatabaseRole=payload_link_reader — the fine-grained access role the job reads the stream through, created by the DDL in syncstorage-spanner/src/schema.ddl. If the role is absent the job fails at startup rather than falling back to broader access.
• pubsubTopic=projects/<PROJECT>/topics/payload-link-changes.

Service account requires:

• roles/spanner.databaseUser on the syncstorage database. This one grant covers both reading the change stream and maintaining the connector’s partition metadata, and in dev the metadata table shares syncdb-dev. Where the metadata database is separate, the grant is needed on both. Note the IAM grant targets the -904c project, so it is applied out of band; see GCP Infrastructure.
• Membership of the payload_link_reader database role, which is what actually narrows the job to the change stream. The IAM grant alone does not let it read BSO rows.
• roles/pubsub.publisher on the destination topic.
• roles/storage.objectAdmin on the Dataflow job bucket, for staging/ and tmp/.
• roles/dataflow.worker at the project level.

2b. Dev/E2E Python publisher — tools/payload-link-dataflow/payload-link-publisher-py/

Plain Python script (not Beam) that polls the READ_payload_link_changes TVF via google.cloud.spanner.Client and publishes the same JSON wire format the Java job produces. Sub-second startup, no JVM. Dev/E2E only — used by the compose stack described below; Java remains the prod publisher.

Follows partition splits: reads _root, picks up child-partition tokens from child_partitions_records, reads each child from its advertised start_timestamp, retires parents once their children are announced, and drops any partition that responds OUT_OF_RANGE. Required in practice — the emulator routes DataChangeRecords to child partitions immediately, so a _root-only reader sees zero DCRs. Full details in the tool’s own README.

3. Pub/Sub topic + DLQ

Provisioned from webservices-infra/sync:

• Topic: payload-link-changes
• Dead-letter topic: payload-link-changes-dlq
• Pull subscription: payload-link-reconciler-sub
  • 60s ack deadline
  • 7d message retention
  • DLQ routing after 5 delivery attempts

4. Reconciler — tools/payload-reconciler/

Python script with one job: pull messages, perform GCS operations, ack. Sync-pull drain loop with two deployment modes selected by whether RUN_BUDGET_SECONDS is set:

• Cronjob mode (default deployment): RUN_BUDGET_SECONDS set (e.g. 240). The script drains the subscription up to that many seconds or until the queue idles, then exits 0. K8s cronjob at ~5 min cadence.
• Long-running mode: RUN_BUDGET_SECONDS unset. The script polls forever, never exiting on idle. Deploy as a K8s Deployment when finalize-flip latency below the cronjob cadence matters.

Per-message handling (reconcile_payload_links.py:handle_message_body):

For each mod in the change record:

• New payload_link present: blob.patch() sets metadata.committed = "true" and customTime = "2200-12-31T23:59:59Z".
• Old payload_link present and not equal to the new value: blob.delete(), with one exception. A batch_bsos row removed under the batch commit transaction tag (transactionTag equals batch_commit, set by syncstorage in the Spanner commit_batch path) is skipped (payload_reconciler.batch_commit_skips). On commit the link moves into the permanent bsos row in the same transaction, so its object must be kept (deleting it was the STOR-657 bug). Any other batch_bsos removal is a genuine delete: batch_bsos has no deletion policy of its own, so TTL expiry and user_collections deletes reach it as cascade deletes with no such tag, and their objects are removed. bsos changes and batch_bsos overwrites are unaffected. See STOR-668.

Both operations tolerate 404 NotFound as success, see Failure modes below.

Environment

Service account requires:

• roles/pubsub.subscriber on payload-link-reconciler-sub.
• roles/storage.objectAdmin on the payload bucket (covers both the metadata patch and the delete operation).

Deployment. Default is a K8s cronjob (~5 min cadence) with RUN_BUDGET_SECONDS set. When lower finalize-flip latency matters, deploy as a K8s Deployment without RUN_BUDGET_SECONDS to run long-running. The cronjob template is sync/k8s/sync/templates/payload-reconciler-cronjob.yaml in webservices-infra, gated on payloadReconciler.enabled. Note it lives in the sync chart rather than sync-jobs.

Four settings in that manifest are load bearing, and changing any of them breaks an assumption the script relies on:

• concurrencyPolicy: Forbid. A run that overruns its window is never joined by a second one. The subscription is the queue, so the next tick simply picks up where the last left off. Two concurrent drains would race on the same messages, which is survivable given idempotency but wastes the ack deadline.
• backoffLimit: 0. No in-window retry. Every operation is idempotent and unacked messages come back on their own, so the next scheduled run is the retry. A backoff would just re-drain the same queue sooner.
• runBudgetSeconds strictly less than activeDeadlineSeconds. The chart fails the render if this is violated. The budget is what makes the drain loop exit cleanly on its own; if the deadline fired first the pod would be killed mid-message and the run would always look failed.
• The command override. The image entrypoint is the syncserver binary, so the cronjob invokes python3 /app/tools/payload-reconciler/reconcile_payload_links.py explicitly. Invoking by path is also what puts the script’s directory on sys.path, which is how its import utils resolves.

Credentials come from workload identity: the pod runs as the tenant GKE service account, which holds the two roles listed above. There is no key file and no secret mounted.

Wire format

Each surviving change record reaches the reconciler as a JSON Pub/Sub message:

{
  "commitTimestamp": "2026-06-30T00:00:00.000000000Z",
  "modType": "UPDATE",
  "tableName": "bsos",
  "transactionTag": "",
  "isSystemTransaction": false,
  "mods": [
    {
      "keys": "{\"fxa_uid\":\"...\",\"fxa_kid\":\"...\",\"collection_id\":1,\"bso_id\":\"...\"}",
      "oldValues": "{\"payload_link\":\"gs://bucket/u/c/b/uuid-1\"}",
      "newValues": "{\"payload_link\":\"gs://bucket/u/c/b/uuid-2\"}"
    }
  ]
}

Mod fields (keys, oldValues, newValues) carry JSON strings that the reconciler parses with a second json.loads, matching Spanner’s change-streams wire convention.

transactionTag carries the transaction tag Spanner records for the change, which the reconciler uses to skip batch commit handoffs (see above). isSystemTransaction is carried too but not currently consumed.

Local e2e compose stack

make docker_run_reconciliation_e2e_tests brings up a full-stack compose environment that exercises the entire pipeline against emulators:

Two publisher variants:

• Python (default) — docker-compose.e2e.reconciliation.yaml. Sub-second startup; the day-to-day iteration path.
• Java (swap-in) — layer docker-compose.e2e.reconciliation.java.yaml on top. Runs the same Java flex-template image under Beam’s DirectRunner. See tools/payload-link-dataflow/README.md for the exact invocation. Use when you need to reproduce a Java-specific issue.

The compose stack doubles as regression coverage for test_storage.py: by opting every collection the storage tests use into GCS offload, each BSO write flows through the offload path (upload + payload_link storage), and each read flows through download_payload. New reconciler-specific tests live in tools/integration_tests/test_payload_link_reconciliation.py, gated by a module-level pytest.mark.skipif on GCS_PAYLOAD_BUCKET so they auto-skip in the existing spanner e2e stack (where the env var is unset) and un-skip here.

Emulator fallback. If a future Spanner emulator upgrade breaks SELECT * FROM READ_payload_link_changes(...), the fall-back is the Java swap-in overlay above (or running the compose stack against a real dev Spanner instance).

Metrics

The reconciler emits statsd counters under the payload_reconciler namespace. Everything the pipeline can tell you about itself is in this list, so it is worth knowing what each one looks like when things are working.

Worth alerting on:

• Anything at all in payload-link-changes-dlq. A message only lands there after five failed deliveries, so it means a record cannot be handled and needs a human. Inspect it through payload-link-changes-dlq-sub.
• errors kind:handler sustained above zero.
• noop_skips climbing, which means the Dataflow filter regressed and the pipeline is paying Pub/Sub for records it discards.
• gcs_404 op:finalize rising as a share of finalizes, which is the signal that objects are being reaped before they get finalized.
• Oldest unacked message age on payload-link-reconciler-sub. This is a Pub/Sub metric rather than one of ours, and it is the most direct measure of finalize latency. It should sit in minutes. The 30 day lifecycle window is the deadline it must never approach.

Failure modes

A payload_link pointing at a bucket other than GCS_PAYLOAD_BUCKET raises ValueError and the message is left unacked — it retries up to the DLQ rather than mutating an unrelated bucket. This is a hard guard.

Keeping the reference patch accurate

tools/payload-link-dataflow/upstream-customization.patch is documentation only — not a build input. If the upstream Cloud_Spanner_Change_Streams_to_PubSub pipeline evolves in ways that change the conceptual diff, refresh it (see the README in that directory). This is a doc refresh, not a code change.