synchronous "sync now" API

[dgreer-dev]

Sync Now: force an immediate commit of a materialization

Changes at a glance

• New Flow capability bit (1 << 21), minted by /authorize/user/task at
  Write+.
• New verified gRPC service on the reactor front door: resolve shard zero,
  self-proxy to its primary, relay to the co-located sidecar. No semantics in
  Go.
• New unary-in/stream-out service in runtime.proto, served by the sidecar
  behind its own capability interceptor.
• Leader FSM: sync-now poke = close_requested + completion waiter, resolved
  at Tail::Done.
• New command: flowctl raw sync-now --task ....
• Open question: restrict to schedule-held transactions, or a general
  "commit what you have" barrier?

Motivation

The runtime sync schedule (#3229) holds a materialization's open transaction
(which continues combining new documents) until the next scheduled commit
instant. That's the right default for cost control, but it conflicts with a
common workflow: "materialize everything you have, then run my analytics." A
user with a nightly-paced warehouse materialization should be able to write

flowctl raw sync-now --task acmeCo/foo/materialize-bar && run-analytics.sh

and know that when the first command exits, everything the materialization had
read at the moment of the call is committed and queryable in the destination.

This ticket delivers the data-plane RPC plus the flowctl raw sync-now
command. This is a runtime-next (V2) feature only, materializations only: V1
tasks are entirely out of scope.

Semantics

Sync-now is a one-shot poke delivered to the task's leader actor. Mechanically
it sets the FSM's existing close_requested input and registers a completion
waiter. close_requested bypasses policy floors (the sync-schedule hold,
configured minimum transaction duration; see the lifecycle-flag rules in
crates/runtime-next/src/leader/close_policy.rs) but never structure:
tail_done gating and unresolved-hint coherence still hold, so a caller
cannot force an incoherent or partial commit. The worst achievable outcome is
more, smaller transactions.

• Open transaction held by the schedule: the hold collapses and it proceeds to
  commit (HELD_COLLAPSED).
• Open transaction already closing: no-op (ALREADY_CLOSING).
• No open transaction: no-op, the task is already current (IDLE).
• Open transaction, no schedule hold in effect: see the open question below.
• Derivations/captures: not applicable (NOT_APPLICABLE); never an error for
  "nothing to do".
• Concurrent and repeated calls are explicitly allowed and must be safe: every
  call is an independent idempotent poke, and N concurrent calls become N
  waiters on the same commit, all resolving together. No rate limiting, no
  queuing, no coalescing machinery. A cron job, a retrying script, and an
  impatient human can all overlap without harm.

Definition of done. The call completes when the transaction that was open
at poke time is fully acknowledged - its contents are committed and visible in
the destination. Data arriving after the poke rides the next transaction and
is out of luck. Done does NOT mean "caught up to the capture's frontier": if
the task is behind on reading, sync-now commits what has been combined so far.
A "wait until caught up to clock X" semantic is a different feature and out of
scope.

Open question: schedule override, or a new screwdriver in the toolbox?

What should sync-now do when the open transaction is not schedule-held -
it's just a big transaction, or one pinned by a configured minimum duration?

Option A - restrict to schedules: sync-now only collapses a
sync-schedule hold; anything else is a no-op with an explanatory outcome. The
feature stays exactly as motivated, and the blast radius is limited to tasks
that opted into scheduling.

Option B - general commit barrier: sync-now means "commit what you have
and tell me when it's queryable", regardless of why the transaction is open
(new outcome CLOSE_REQUESTED).

RPC shape: server-streaming with two messages

The RPC streams exactly two messages:

1. Ack (immediate): the outcome enum - HELD_COLLAPSED | ALREADY_CLOSING | IDLE | NOT_APPLICABLE, plus CLOSE_REQUESTED under Option B - and a
   snapshot of the open transaction's extents (sourced docs, bytes, open age),
   which the FSM already tracks. Immediate confirmation of what the call is
   waiting on.
2. Done (on completion): committed stats for the transaction. Sent when
   the leader FSM's Tail reaches Done for the target clock recorded at
   poke time.

A caller that doesn't want to wait just hangs up after the ack; the poke has
already landed server-side, so early hangup is harmless. This avoids two
server-side code paths for wait/no-wait.

Dependency: connector-side ack schedules must not stack

This is the one external blocker, and it needs a decision. Eleven
warehouse-family connectors (snowflake, bigquery, databricks, redshift,
motherduck, clickhouse, fabric-warehouse, both icebergs, filesink) implement
the legacy sync schedule by delaying their Acknowledged response - and it
is default-on: an absent connector syncSchedule means a 30-minute fixed
ack delay (defaultSyncFrequency in
connectors/go/materialize/schedule_config.go). That delay lives inside the
connector, beyond the runtime's reach. Sync-now can collapse the runtime hold,
but the commit then sits in the connector's ack delay for up to 30 minutes:
the wait stays correct (it resolves on true completion), but the caller's
script blocks on a delay the runtime cannot collapse, which defeats the
feature.

The build-time SyncScheduleConflict check rejects a configured connector
schedule alongside a model schedule, but today:

• absent connector config + model schedule passes validation and silently
  stacks with the 30m default;
• the connector's documented "off" state (syncFrequency: "0s") is counted as
  configured and rejected - so there is currently no expressible config
  state that yields runtime-only pacing on these connectors.

Proposal:

1. Validator carve-out: treat an all-zero connector schedule
   (syncFrequency: "0s", no fast-sync fields) as unconfigured in
   connector_config_has_sync_schedule. Explicit 0s becomes the sanctioned
   pairing and works on every connector image already shipped.
2. Gate model syncSchedule on the V2 runtime flag at validation. Today a
   V1 task can carry a model schedule that is silently inert; after the
   deferral change below it would be actively harmful (no pacing at all).
3. Boilerplate deferral: the connector receives the built
   MaterializationSpec, which now carries sync_schedule_json; change
   shared boilerplate so a spec-level schedule forces the connector ack
   schedule off (config and default both). One change covers all eleven
   connectors; V1 tasks are untouched. Ships on the normal image cadence;
   explicit 0s is the bridge until a task's image has it.
4. Eventual removal of the connector-side machinery once conversion is
   complete and a census shows no remaining users; the config field stays as a
   tolerated no-op for stored-config compatibility.

If the open question resolves to Option B, this dependency has a second face:
an unscheduled warehouse task has no model schedule, so the deferral never
triggers and the connector's default 30m ack delay is in play - a sync-now
wait on such a task can legitimately block on it. Correct, never early, but
slow. The mitigation is the same explicit syncFrequency: "0s" (no conflict
arises without a model schedule); document it.

Sync-now itself builds no detection machinery for this: it's a documented
dependency, not a runtime check.

Failure modes

• No live leader session (task disabled, V1 task, mid-restart, shard zero
  relocated): sidecar returns NOT_FOUND; the Go handler surfaces it; flowctl
  retries Shard.List once to catch a relocation, then errors with a hint
  that the task may not be running on the V2 runtime.
• Session dies mid-wait (shard reassignment, leader restart): parked
  waiters get an error; re-invoking is idempotent and re-establishes the
  contract ("all data as of the retry").

flowctl

flowctl raw sync-now --task acmeCo/foo/materialize-bar, reusing the
TaskSelector and user_task_authorization plumbing from raw list-shards.
Waits for done by default (the script use case); a --no-wait flag hangs up
after the ack.

Out of scope / follow-on

• A dashboard "Sync Now" affordance calling this same RPC. Note for that work:
  browsers can't speak raw gRPC, so the UI needs a grpc-web/CORS layer on the
  reactor path or a proxy; recorded here so the dashboard team knows the gap.
• Derivations (coherent under Option B; natural follow-on if wanted).
• Any V1-runtime behavior.
• "Caught up to frontier" wait semantics.
• Hold-suppression windows, "skip next hold" arming, rate limiting.

[williamhbaker]

Generally looks good, a few notes based on external conversations & maybe one or two additional thoughts:

• I'd base the auth capability on the existing read bit for now, unless some finer-grain requirement comes up later
• For the schedule override vs. general mechanism: General mechanism. This will likely be used exclusively when a sync schedule is in place, but I don't think there's any reason to limit it to that, and it wouldn't surprise me if it ends up being an easier implementation to make it general
• The connector-side ack schedules taken as-is are a complication, and a connector configured with one won't work as expected with this sync now mechanism. I don't think we need to do anything more with this though, a user who wants to use "sync now" will just have to move their scheduling out of the connector and into the runtime-level config...or we'll have done it for them via an automated migration
• The UI button path to activating this will be at least kinda important, so good to think about that up front.
• On the efficiency / service protection angle: ...every call is an independent idempotent poke, and N concurrent calls become N
  waiters on the same commit... - maybe we could take this even further, and have subsequent calls actually replace and cancel prior ones, that way we are only ever tracking a single request