[Proposal] Scalable replication workers

[rkistner]

Background

Currently, we recommend running exactly one replication process. This process is responsible for replicating both "active" and "reprocessing" sync rules, doing both streaming replication and snapshot replication for each.

This has various issues:

1. Performance on active sync rules can degrade while reprocessing with sync rules, due to increased CPU usage, memory usage and lock contention.
2. Concurrency support is limited - it cannot process snapshots while streaming, and can only process one snapshot at a time. And if we change this, it further adds to the overhead on that process which can further degrade performance on active sync rule processing.
3. It does not help to run multiple replication workers in most cases - only one will perform work at a time.

This work forms part of the larger Incremental reprocessing project.

Proposal

The basic idea is to allow multiple processes to do replication concurrently, with a configurable number of "workers" on each.

New config option:

replication:
   workers_per_process: 1 # default of 2, to match current behavior.

This configures the number of workers in each process. By specifying the max as 1 and running multiple processes, we avoid the performance issues from concurrent processing within one process.

Additionally, the CLI "role" option will be expanded to get two new sub-options:

1. --role replication (same as the current --role sync, just a rename to be less ambiguous)
2. --role replication:streaming
3. --role replication:snapshot

This can restrict the worker(s) in a process to only do streaming or snapshot replication, instead of both.

Auto-scaling

The metrics produced by replication workers will be expanded to include some new ones:

powersync_total_stream_count
powersync_active_stream_count
powersync_pending_snapshot_count
powersync_active_snapshot_count

This can then be used with HPA config for auto-scaling:

1. powersync run -r replication:streaming, min replicas = 1, target replicas = powersync_total_stream_count, max replicas = 2.
2. powersync run -r replication:snapshot, min replicas = 0, target replicas = powersync_pending_snapshot_count, max replicas = 4.

This config will run a single streaming replication process in the steady state, but up to 5 additional processes to handle re-replication and snapshots.

Internal concurrency improvements

To take full advantage of this, we need some internal improvements:

1. Consider using a worker_thread per worker to remove the CPU as a bottleneck for multiple workers per process.
2. Pre-generate op-id sequence number batches for each worker, to allow flushing independent changes concurrently.
   a. This one could be tricky - we'll need to thoroughly investigate consistency guarantees.

Deployment architecture

While the new options give flexibility in how the replication process(es) are deployed, we'll have a couple of recommended setups:

1. Auto-scaling: Described above. This gives a great combination of resource efficiency and performance by utilizing Kubernetes HPAs or similar auto-scaling tooling.
2. Single large process: powersync run -r replication with replication.workers_per_process = 4, and generous CPU and memory limits (say cpu: 2 and memory: 2Gi). This simplifies the deployment compared to auto-scaling, at the cost of more resource usage.
3. Single small process: powersync run -r replication with replication.workers_per_process = 2, and smaller CPU and memory limits. This is effectively the same setup as before these changes, and could still be a good choice for development setups.
4. Multiple processes, no auto-scaling: 3-4 processes with powersync run -r replication, replication.workers_per_process = 1. This is similar to the single large process setup, but gives some redundancy.

Processing priorities

Work is prioritized in this order:

1. Snapshots for active sync rules.
2. Stream for active sync rules.
3. Snapshots for reprocessing sync rules.
4. Stream for reprocessing sync rules.

Note that:

1. A minimum of 2 workers is required in total to support reprocessing sync rules, since the active sync rules stream never completes, and always takes higher priority than reprocessing.
2. With 2 workers, if the active sync rules need to re-snapshot tables, that takes higher priority than reprocessing sync rules.
3. We may need to interrupt existing work if higher-priority work comes in.

Relation to incremental reprocessing

The changes here can be implemented independently, but are also part of the larger Incremental reprocessing project. That project affects this in some ways:

1. Right now, replication for snapshots cannot be performed independently from streaming replication. This changes with [Incremental Reprocessing] [WIP] [Postgres] stream while snapshotting #426, and the config changes here is a requirement to roll out those changes.
2. There will eventually only be one replication stream, which will cover both active and reprocessing sync rules. While this does not affect the config options directly, it may affect the recommended production config.

[henriquekraemer]

We've been hitting the "only one snapshot at a time" limitation from a slightly different angle, and built a working prototype of the internal concurrency pieces from this proposal, specifically for Postgres source + Postgres storage on storage v2. Sharing the data here since it speaks directly to a couple of the open points, particularly the op-id / consistency question.

Context: with wildcard schemas (#703), one sync config resolves to hundreds of source tables (tables x tenant schemas). On our deployment (200+ tenant schemas), initial replication snapshots them strictly sequentially, and a full re-replication takes hours. The "stream while snapshotting" work landed for MongoDB (#641), but the Postgres side (#426) was closed, so this path is still open here.

Where the time actually goes

Instrumented WalStream.snapshotTable and PostgresBucketBatch.replicateBatch. Steady state for a 10k row chunk (~530B rows, ~780ms), Postgres storage:

• source read (cursor stream wait): ~50ms (6%)
• decode + applyRowContext: ~30ms (4%)
• current_data existence lookup (skipExistingRows): ~82ms (10%)
• bucket eval + op build (saveOperation): ~135ms (17%)
• storage write (the INSERTs, one connection, serialized): ~540ms (68%)

[storage-timing] flush 10000 ops: total 780ms | current_data lookup 82ms | eval+build 134ms | write 547ms
[storage-timing] flush 10000 ops: total 773ms | current_data lookup 81ms | eval+build 129ms | write 542ms

Raising batch_limits.max_record_count from 2000 to 10000 changed nothing measurable, so the write path is not round-trip bound, it is just serialized. This is why concurrent flushing (the op-id batching item below) is the piece that matters most for Postgres.

Prototype

Branch: henriquekraemer/powersync-service@feat/snapshot-concurrency (module-postgres and module-postgres-storage suites pass). It implements two of the internal concurrency improvements from the proposal, scoped to the Postgres initial snapshot:

• Concurrent table snapshots. N workers (POWERSYNC_SNAPSHOT_CONCURRENCY, default 1 = current behavior), each with its own snapshotConnection() and its own storage writers via createWriter(), pulling tables from a shared queue.
• Pipelined flushes within a table. Two alternating writers per worker: chunk N's flush runs while chunk N+1 is read and evaluated. updateTableProgress is chained strictly in chunk order, after that chunk's flush and all earlier ones, so resume semantics are unchanged.

This leans on two existing properties, which I think are the reason it works on v2 without touching the consistency model: each table records its own consistency LSN (markTableSnapshotDone(tableLsnNotBefore)), so there is no cross-table point-in-time requirement (WAL replay reconciles); and chunked snapshots cover disjoint PK ranges with per-table resumable progress.

Results on a small staging host, 4 workers:

• baseline sequential: ~11.5k rows/s
• prototype: ~18k rows/s (~1.55x)

16:57:36.830Z Replicating "tenant_b"."maint_order_notes" 5480000/~7880678 [snapshot-timing] pipelined read+eval 784ms (stream-wait 665ms)
16:57:38.839Z Replicating "tenant_a"."maint_order_worktimes" 620000/~2490426 [snapshot-timing] pipelined read+eval 188ms (stream-wait 34ms)
16:57:40.072Z Replicating "tenant_a"."maint_order_notes" 29190000/~29106792 [snapshot-timing] pipelined read+eval 1420ms (stream-wait 1318ms)
16:57:41.908Z Replicating "tenant_b"."maint_order_worktimes" 120000/~479605 [snapshot-timing] pipelined read+eval 1332ms (stream-wait 1167ms)

On the points from the proposal

"Consider using a worker_thread per worker to remove the CPU as a bottleneck." Confirmed on our host. Above ~1.55x the ceiling is the Node main thread, pegged at ~99% of one core (decode, bucket eval and wire serialization share it). The inflated stream-wait values above are event loop queueing, not a slow source. So the CPU bottleneck is real and single-process concurrency alone tops out fairly quickly; worker_threads (or the multi-process model in this proposal) is what unlocks the rest.

"Pre-generate op-id sequence number batches for each worker... we'll need to thoroughly investigate consistency guarantees." In the prototype the workers do not pre-allocate op-id ranges; each writer just calls nextval('op_id_sequence') at flush time, so op-ids interleave across writers rather than being contiguous per table. Across full re-replication runs we saw no correctness issues from this, since bucket_data is append-only and each table gates on its own LSN. That may or may not hold for the streaming case you have in mind, so I'd like to understand whether contiguous per-worker op-id ranges are a hard requirement or an optimization. This is the main thing I'd want your read on before building further.

One concrete bug found along the way. clearError() runs inside the REPEATABLE READ replication transaction on a writer's first flush, while every flush of every writer updates the same sync_rules row via the keepalive in withReplicationTransaction's finally block. With concurrent writers this produced 40001 retry storms, and each retry redoes the whole flush. Moving the clearError to after commit (plain autocommit update, like the keepalive) gave zero serialization failures across full runs. This is independent of everything above; happy to send it as a standalone PR.

One more, orthogonal

The skipExistingRows lookup queries current_data for all 10k keys on every flush. Since chunks cover disjoint PK ranges, every row after the resume-boundary chunk is guaranteed new, so the lookup is only needed on the first chunk after a restart. On a ~38M row current_data we measured 5 to 6 seconds per flush on cold key ranges:

[storage-timing] flush 10000 ops: total 6973ms | current_data lookup 5213ms | eval+build 136ms | write 1605ms

An opt-in hint (e.g. assumeNewRows on the writer or save operation) would remove that. Worth a separate issue?

Happy to adapt any of this to fit how you'd like the Postgres side to land relative to the multi-process model here. Tenant schemas are renamed in the logs above.

[henriquekraemer]

Thanks! That timeline makes sense for the full multi-process model. One question on the interim, though: the #349 progress matrix has Postgres pending on both the source and storage side, so for Postgres users the v4 concurrency work is realistically some way out, and full re-replication is painful today (some of the reports in #349 are exactly that). Would you be open to landing the in-process version as an experimental, config-gated option on v2 in the meantime?

Concretely:

• Default stays 1, which is exactly the current sequential behavior and resource profile. Nothing changes for anyone who does not opt in.
• Scoped to the Postgres initial snapshot only; no protocol or storage schema changes, and the module-postgres and module-postgres-storage suites pass.
• We run it on our own deployment, and we'd maintain it (rebases, fixes) until the v4 model replaces it. I'm aware the v4 Postgres port will rework this same area, and [Proposal] Bucket data on object storage #376 may change the write-path economics entirely, so flagging it as experimental seems right.
• The one design question I'd want your read on first is the op-id one from my earlier comment: whether interleaved (non-contiguous) op-ids across concurrent snapshot writers are acceptable on v2, given bucket_data is append-only and each table gates on its own LSN. If yes, the rest is contained.

And totally fine if you'd rather keep this area frozen until v4; the branch works for us either way, and I'll keep it rebased.

On the resource usage question, some numbers from running the prototype on deliberately small hardware (2 vCPU app host): the config-gated approach sidesteps part of the problem, since users who opt in size the process for it, same as any other tuning knob. On our host the useful range turned out to be small anyway: 2 workers already saturate the storage tier, and above ~1.55x the Node main thread caps out. So in-process concurrency mostly limits itself, and the multi-process model from this proposal is what deployments that need more would reach for.

One operational note that applies to both models: the binding resource moved around depending on environment. On the app host it was the single core; once that was fine, it was storage write throughput. When storage got slow (we tested against a degraded storage instance), pipelining absorbed the backpressure gracefully, but chunk processing time eventually exceeded the fixed 30s snapshot socket timeout on the source connection, which killed the replication and caused a restart loop that re-read the same chunk against an already-slow storage. Whatever concurrency model lands, making the source snapshot connection resilient to slow storage (keepalive during flush, or a configurable timeout) is probably a prerequisite, since concurrency makes storage saturation easier to reach.

Separately, three smaller items independent of all of the above:

1. The clearError() serialization fix described above. Want a PR for that one?
2. assumeNewRows to skip the current_data existence lookup during initial snapshot (the 5 to 6s cold lookups). Happy to file it as an issue.
3. Related scale issue we hit terminating an old sync rules group: PostgresSyncRulesStorage.clear() deletes the whole group in single unchunked DELETEs. With a 79M row current_data group the DELETE never completes before the connection drops, retriedQuery gives up after 2 attempts, terminate never reaches TERMINATED, and the replicator retry loop re-issues the giant DELETE forever (storage bloat plus wasted load, though the active stream is unaffected). The snapshot truncate path already batches with BATCH_LIMIT 2000; clear() could do the same. Can file this as an issue too.

[rkistner]

Would you be open to landing the in-process version as an experimental, config-gated option on v2 in the meantime?

Right now our bottleneck is to a large extent in the review and testing side, as well as in managing merge conflicts between overlapping storage changes, so I'm not confident we'll be able to land it quickly. Regardless, if you open a PR, it can help us to speed up the implementation, whether we use your PR as-is or not. It could also help to more clearly evaluate the effort that would be involved in landing the changes.

The one design question I'd want your read on first is the op-id one from my earlier comment: whether interleaved (non-contiguous) op-ids across concurrent snapshot writers are acceptable on v2, given bucket_data is append-only and each table gates on its own LSN. If yes, the rest is contained.

That sounds fine for Postgres storage, I can't think of any issues. I believe my original comment was more in relation to MongoDB storage, which does not have native support for auto-incrementing ids, which may make the implementation more complex.

The clearError() serialization fix described above. Want a PR for that one?

Yes, a PR for that would be great.

assumeNewRows to skip the current_data existence lookup during initial snapshot (the 5 to 6s cold lookups). Happy to file it as an issue.
The skipExistingRows lookup queries current_data for all 10k keys on every flush. Since chunks cover disjoint PK ranges, every row after the resume-boundary chunk is guaranteed new, so the lookup is only needed on the first chunk after a restart.

In principle in makes sense that in the majority of cases, all rows will be new, and we don't need to lookup the existing ones. The main thing I'm concerned about here is covering all edge cases: We currently treat the operations as idempotent, and there may be more cases relying on that. "the first chunk after a restart" is one specific example, I'm just concerned that there may be more cases like that. As a slightly contrived example, if the batch size is reduced at the same time as restarting the process during a snapshot, it could affect more than just the first chunk.

So I'm wondering if there's a more robust way we could avoid the performance overhead for the common case, but still handle existing rows gracefully if they do exist? One option could be to eagerly attempt the save, but if it runs into duplicate key errors, retry with a lookup. Would something like that work? It may even be possible to use that optimization for all inserts, without needing an explicit assumeNewRows flag.

Aside from that, I do find it strange that the current_data lookup is so slow in your example - I can't imagine why it would take 5s to lookup the (non-existent) rows, then only 1.6s to write them. Maybe there's a further performance issue with the lookup query?

Related scale issue we hit terminating an old sync rules group

When we redo the Postgres storage, this should be eliminated complete by using partitioned tables, just dropping the entire table instead of deleting individual rows. But it may be worth implementing the batch limit in the meantime.

[henriquekraemer]

The op-id answer unblocks the main design question then. clearError PR is up: #726.

One option could be to eagerly attempt the save, but if it runs into duplicate key errors, retry with a lookup. Would something like that work? It may even be possible to use that optimization for all inserts, without needing an explicit assumeNewRows flag.

Yes, and on Postgres storage it can skip even the retry. The current_data flush is already INSERT ... ON CONFLICT DO UPDATE; for snapshot inserts it can become ON CONFLICT DO NOTHING ... RETURNING, so a single statement inserts and reports which rows already existed, and bucket_data ops only get emitted for the returned (new) rows. Existing rows get detected at write time, so the resume-boundary case, the batch-size-change case and whatever else are all covered without a flag. The wrinkle is ordering: bucket_data ops are currently built before the current_data flush, so emission has to move after the write, but that stays contained in module-postgres-storage. On MongoDB storage the insert-then-retry-on-duplicate version you describe would be the equivalent.

Aside from that, I do find it strange that the current_data lookup is so slow in your example - I can't imagine why it would take 5s to lookup the (non-existent) rows, then only 1.6s to write them.

Fair, and after re-measuring I owe a correction there: those numbers came from a degraded storage instance. It still had a 79M row orphaned group from the clear() issue, plus heavy dead-tuple bloat from its retried DELETEs. On clean storage the same lookup is 30 to 600ms per 10k flush. So no separate bug in the lookup query; skipping it still saves the steady-state 10 to 20%, but the 5 to 6s was the environment, not the query.

But it may be worth implementing the batch limit in the meantime.

Agreed, I'll file an issue for that. Partitioned tables in the v4 storage sounds like the right end state.

I'll also get the snapshot concurrency PR into shape (proper config option instead of the env var, tests for the concurrent path) and open it, understanding it may sit for a while given your review bandwidth.

[henriquekraemer]

Opened the PR: #731. It has the pipelined chunk flushes and the snapshot_concurrency option, with the staging benchmark numbers in the description.

Writing a test for it with a small chunk size also caught a checkpoint coverage issue with multiple writers (ops flushed by a secondary writer could stay above last_checkpoint until unrelated streamed changes advance it). The fix and the reasoning are in the PR description.

[henriquekraemer]

The batched clear() fix is open too: #734. Together with the snapshot concurrency PR, that covers everything I had pending from this thread.