Runtime "Sync Now" - #3326
Open
dgreer-dev wants to merge 3 commits into
Open
Conversation
williamhbaker
left a comment
williamhbaker
left a comment
Member
There was a problem hiding this comment.
Some directional comments! I didn't get into the gritty details just yet since I am hoping things can be simplified quite a bit.
dgreer-dev
marked this pull request as draft
August 10, 2026 23:37
Contributor
Author
|
Pushed a commit with the four resolved convo fixes.
|
Remove the check which rejected a materialization configuring a connector-side `syncSchedule` alongside a model-level one, and with it `connector_config_has_sync_schedule` -- which sniffed the undecrypted endpoint config for the key, treating an empty object or all-null / all empty-string values as absent so that a UI clearing the schedule didn't leave a tripwire behind. Configuring both is not something we want users to do, but it doesn't break: the two pacing mechanisms just interleave chaotically, and we plan to migrate extant configs automatically. That isn't worth the complexity, nor the guesswork of deciding what a disabled connector schedule looks like. It also stood in the way of runtime-only commit pacing, where a model-level `syncSchedule` is the sole cadence control and the connector's is left at its documented all-zero off switch. The connector parses a Go duration, which accepts "0m0s" and a bare "0" as readily as "0s", so the check would have had to enumerate zero spellings to let that through. `SyncScheduleInvalid` stays: a malformed schedule is still an error.
…leader Sync-now forces an immediate commit of a materialization's open transaction -- collapsing any sync-schedule hold -- and resolves once that transaction is fully acknowledged (committed and queryable in the destination). Define it as a new TaskControl gRPC service in runtime.proto, kept separate from Leader because its callers present ordinary gazette READ claims over the task's shards rather than the LEAD capability. The response stream is exactly one Ack, then zero or more Heartbeats, then exactly one Done. Its messages are structural and carry no payload: transaction statistics are recorded to the task's stats journal, which is where callers read them. There is no error for "nothing to do" -- a task with nothing to await Acks and its Done follows immediately. - The TaskControl tonic service registers beside Leader on the sidecar port, behind a gazette READ AuthN floor rather than LEAD. A live session is reachable through a per-session SyncNowHandle (task name -> delivery channel plus shard-zero ID, un-registered by guard with the session); the caller's claims are authorized against shard zero and the request is delivered into the Actor's select loop. Anything not addressable that way is NOT_FOUND, including a capture or derivation: only the reactor front door can distinguish those, because only it has the shard keyspace to resolve a task's type. - The Actor decides what a request awaits with a pure function over POD FSM state (sync_now::evaluate over fsm::sync_now_inputs): one Tail::Done per pipeline stage ahead of the caller, and the existing `close_requested` input armed when an open transaction's close decision is still ahead. - Waiters park on a target count of Tail::Done transitions -- the "committed and queryable" instant -- so N concurrent requests await the same commit. The Begin->Done stopping shortcut is excluded from that count: it defers connector acknowledgement to a future session rather than completing it. Parked waiters receive ~15s Heartbeats via the actor's existing wake machinery. - A session exiting with parked waiters errors them Unavailable, and closes the request channel before its final drain so that a request racing teardown is likewise answered Unavailable rather than accepted and discarded. Sync-now is idempotent and callers re-invoke. An e2e test pins the contract on the Rust side: a real tonic server hosting Leader + TaskControl behind armed AuthN interceptors (assembled exactly as the runtime sidecar serves them), a real materialize leader session, and real journal IO for the stats and ACK-intent writes that gate Tail::Done. The test plays shard zero itself over the wire, with a scripted connector side whose withheld Acknowledged is its control over when a transaction becomes fully acknowledged. Source checkpoints come from a fixture ShuffleSessionFactory fed synthetic Frontiers, so no journals are read. Fixture tasks cover both hold shapes: a 2h syncSchedule and a 300s min-transaction-duration floor. Covered: a held transaction collapses on SyncNow and Done arrives only after the shard-side ack, with an inline re-poke; N concurrent pokes resolve at one shared commit; a poke in the post-commit drain window awaits the same ack; a close request bypasses the min-duration floor; parked waiters receive Heartbeats (the one deliberately slow test -- the 15s cadence has no test knob on purpose); token scope and capability enforcement plus NOT_FOUND and InvalidArgument resolution over the real wire, including a capture, which this service answers NOT_FOUND because task types are resolved by the reactor front door; and a session exiting with parked waiters errors them Unavailable and un-registers its handle. A schedule hold is proven by the leader's silence -- absent a poke it asks the shard for nothing -- and the drain barrier by Done itself, which is what proves Tail::Done was reached. Held-transaction tests warm up with a first transaction (never schedule-held, by session_start) and guard the ~1-in-1e5 wall-clock flake of opening adjacent to a schedule grid instant with one bounded retry. Adds a rustls dev-dependency to install the process-level CryptoProvider that production mains install; pinned to 0.23 because the workspace's rustls = "0" re-resolves to the also-locked 0.21 (which lacks the aws_lc_rs feature) when named from this crate.
Expose sync-now to users: the reactor front door relays TaskControl to the
sidecar hosting a task's leader, over a hand-written REST/NDJSON endpoint
which both the dashboard and flowctl call. gRPC is used only on the
internal front-door-to-sidecar hop; the front door serves no TaskControl
gRPC of its own.
The relay core resolves the task's shard zero -- found by item-prefix scan
over `<task-type>/<task-name>/` with an exact task-name label match, since
one task's name may prefix another's -- through the consumer framework's
resolver with the caller's verified claims, so the token's label selector
is enforced against the task's shards (AuthN floor is gazette READ; no new
capability bit). The resolution is for routing and AuthZ only, and is
released immediately after: an hour-long relay must not pin the shard's
teardown, and a mid-relay primary move just breaks the stream (SyncNow is
idempotent; callers re-invoke). Any front door can then dial any sidecar
directly -- they listen on a fleet-wide port, and it's the same cross-host
sidecar dial materialize_v2.go already makes for LeaderEndpoint -- so
there's no gazette-style proxy hop to the primary's own front door. The
caller's Authorization is forwarded verbatim for the sidecar to
independently verify.
One consequence of dialing the sidecar rather than proxying to the
primary's front door: a non-primary front door doesn't inherit the
primary's storeReadyCh wait, so a sync-now landing during shard recovery
is likelier to see NOT_FOUND. That wait was never a real guarantee --
store-ready does not imply the sidecar's leader session has reached Join
consensus, which is why flowctl retries a post-Ack NOT_FOUND at all.
Because it resolved the shard, the front door also knows the task's type,
and answers a capture or derivation itself rather than dialing the sidecar
-- which sees only a task name. That keeps the contract uniform across
task types: "not running here" is NOT_FOUND whatever the task is, and a
task which simply holds no transaction to force Acks and is done.
The HTTP endpoint `/v1/task-control/sync-now` mounts beside gazette's
grpc-gateway `/v1/` mux with the same CORS wrapper, and reproduces the
gateway's streaming dialect exactly -- `{"result": ...}` NDJSON lines via
gogo/gateway's JSONPb with EmitDefaults, flushed per message, and terminal
`{"error": {grpcCode, httpCode, ...}}` lines -- so the UI's existing
stream parsing needs nothing new. The endpoint's doc comment is the UI
contract for the dashboard's "Sync Now" button. Both deps were already
pinned transitively and are promoted to direct.
A Go test stands up two consumertest members over etcdtest with a stub
sidecar TaskControl server, covering relay from the shard's own primary
and from a front door which is not it, task name and bearer token arriving
intact, claim rejection (missing / mis-scoped), NOT_FOUND passthrough, a
capture answered without the sidecar being dialed at all, and a
gated-flush assertion that each NDJSON line is readable before the stream
completes.
flowctl: `flowctl raw sync-now --task acmeCo/foo/materialize-bar` exits
once the awaited transaction is fully acknowledged, so that `flowctl raw
sync-now --task X && run-analytics.sh` sees the freshest data. It prints
nothing: the exit status is the whole contract. Authorization is the
ordinary Read-capability /authorize/user/task token every task reader
already holds; no capability changes anywhere. dataplane.rs gains the pair
`user_task_auth_watch` (starts the watch) and `reactor_front_door` (reads
an address and token from it), because the caller must hold the watch: an
hour-long wait can outlive its reactor token, and each attempt re-reads
rather than reusing. The watch is now the single construction site for
task authorizations in this module.
The client reads SSL_CERT_FILE explicitly, because reqwest verifies
against webpki's bundled roots rather than the OS roots that tonic's
tls-native-roots uses; without it every private-CA data plane, including
local stacks, would be unreachable.
Failures are classified by gRPC code, not by how far the stream got.
Unavailable is always a leader transient: the leader itself reports it when
a session ends before its awaited transaction is acknowledged, and when a
request races session teardown. DeadlineExceeded is the claims deadline
which gazette's KeyedAuth.Verify binds the relay to -- reactor tokens live
for a jittered 40 to 80 minutes, so every wait outliving one dies this way,
and the next attempt's freshly-read token resolves it whether or not we had
been acked. NOT_FOUND and EOF-before-Ack are the "not running here, or not
on the V2 runtime" diagnostic only until some attempt has been
acknowledged; afterwards they mean a leader restarted and its replacement
is not addressable until it reaches Join consensus, which is exactly the
state a reconnect lands in. So they are fatal on a first attempt and
retryable after one. EOF before Done is never success, since a clean exit 0
without a Done would silently break the shell contract above. An unparsed
response (a proxy or load balancer answering in the data plane's stead) is
retryable iff 5xx, and a failed request or aborted body always is.
Retries back off from 1s to 30s and continue indefinitely, because a caller
who wants a deadline can bound the invocation itself and one who doesn't
wants the sync.
Seven tests drive the response handling against a stub front door over a
real loopback socket: Done-waiting through heartbeats, a table over the
retry classification, a mid-stream body abort, EOF before and after an Ack,
a terminal error line, and an unparsed gateway response. The retry loop
itself is not covered, as it needs a control-plane authorization watch.
dgreer-dev
force-pushed
the
daveg/3281-sync-now
branch
from
August 12, 2026 15:37
731ba0e to
c71883f
Compare
dgreer-dev
marked this pull request as ready for review
August 12, 2026 15:37
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
flowctl raw sync-now --task X(andPOST /v1/task-control/sync-now) forces animmediate commit of a V2 materialization's open transaction and returns once that
transaction is committed and queryable in the destination. Closes #3281.
The workflow we want to enable is a user wanting to run analytics on the freshest
data available, regardless of any sync schedule (or other transaction-holding
reason). Essentially we want to allow
flowctl raw sync-now ... && ./my-analyitcs.pyto be a safe and useful command. See below for specifics of the new command.
Design points worth your attention
it's queryable", whatever is holding the transaction (schedule, min-duration
floor, or just size). Worst case is a smaller transaction. "Nothing to do" is
never an error: an idle task Acks
IDLE, a capture or derivationNOT_APPLICABLE, both exit 0.IDLE/NOT_APPLICABLEresolves at thetarget transaction's
Tail::Done, so N concurrent callers park on one commit.zero with the caller's claims, proxies to the primary's peer via
mayProxy,and relays 1:1 to the co-located sidecar.
/v1/task-control/sync-nowexists as a secondway to force the sync. Calls the same code as flowctl, just an HTTP different path.
connector
syncSchedulecounts as unconfigured, so runtime-only pacing isexpressible on every connector image already shipped; and a model-level
syncSchedulewithoutenable-runtime-v2is now rejected rather thansilently inert.
Verification
Rust e2e tests ensure transaction control works as advertised. Go tests stand up
two consumer test members against a stub sidecar, covering both transports
including the peer-proxied path. flowctl's response handling is covered against a
stub server over a real channel.
Also verified on a local stack against a schedule-held V2 materialization
(transcripts available). Highlights: three parallel
flowctlpokes plus onecurl -Nstream against one held transaction all exited 0 reporting identicalcommitted stats (one
HELD_COLLAPSED, threeALREADY_CLOSINGwaiters);destination row count equalled the pre-poke source count exactly on every run.
Example Usage
Ack is immediate and carries the outcome plus a snapshot of the awaited
transaction. Done arrives when that transaction is committed and queryable in
the destination, with its committed stats.
durationMillisis thetransaction's total open age, not the commit duration. Progress heartbeats
(~15s) are consumed for liveness but not printed unless
--progress, since anhour of them is a lot of lines unasked.
Any completed outcome exits 0,
IDLEandNOT_APPLICABLEincluded - "nothingto do" is success. Exit is non-zero for
NOT_FOUND(not running here, or noton the V2 runtime) and for
--timeoutelapsing.Flags:
--progressprints the heartbeats;--no-waitprints the Ack and hangsup, which is harmless because the commit has already been forced;
--timeout 10mbounds the whole invocation, which otherwise waits indefinitely and retriesleader restarts.
The same call over the REST transport, which is what the dashboard button will
use:
Commits