diff --git a/doc/developer/design/20260825_durable_subscribe.md b/doc/developer/design/20260825_durable_subscribe.md new file mode 100644 index 0000000000000..b34cbee3bd5d9 --- /dev/null +++ b/doc/developer/design/20260825_durable_subscribe.md @@ -0,0 +1,932 @@ +# Durable subscribe + +- Associated: no tracking issue filed yet, see "Open questions". + +## The Problem + +`SUBSCRIBE` cannot be resumed. A client that loses its connection has no way to +continue from where it stopped, so it must re-run the subscribe and process a +fresh snapshot before it sees a single new update. The cost falls on exactly the +clients least able to absorb it, because a browser tab or an edge function +reconnects often and holds no durable local state to fall back on. This makes +reconnection a data-volume event rather than a control-plane event. + +The gap is not a missing streaming mechanism. `SUBSCRIBE ... AS OF t WITH +(SNAPSHOT = false)` already expresses "give me the diffs after `t`". What is +missing is any guarantee that `t` remains readable: nothing holds the +collection's `since` on the consumer's behalf, and the default compaction window +is one second (`DEFAULT_LOGICAL_COMPACTION_WINDOW_MILLIS` in +`src/adapter-types/src/compaction.rs`). A client that remembers a timestamp and +comes back a second later finds it already compacted away. + +Our own console demonstrates the workaround and its cost. +`console/src/api/materialize/SubscribeManager.ts` opens one dedicated WebSocket +per subscribe, and on reconnect it re-runs the subscribe from scratch, holding +the stale snapshot behind a `resubscribing` flag until a new snapshot completes. +The manual escape is documented in +`doc/user/content/transform-data/patterns/durable-subscriptions.md`, a shipped +private-preview page that instructs users to set a `RETAIN HISTORY` duration, +record progress timestamps themselves, and resume with `AS OF +`. Every part of that is a responsibility we are +asking the client to carry, including the off-by-one. That page was titled +"Durable subscriptions", which this design renames to "Resuming subscriptions" to +free the name for the object. + +A second problem sits underneath the first. `SUBSCRIBE` has no persisted output, +so it has no reconciliation point of the kind that makes materialized views +robust. The materialized view sink is self-correcting because it reads its own +output shard back through persist feedback into a `Correction` buffer +(`src/compute/src/sink/correction_v2.rs`) and writes only the delta needed to +make the shard match the dataflow's desired output, which is what survives +restarts and replica changes. Resume correctness for a subscribe would otherwise +rest on the client's unverifiable claim about what it holds. + +## Success Criteria + +* A client that reconnects within an agreed window continues from where it + stopped, with no new snapshot and no gap. +* The server owns the resume point. A client is not required to durably remember + a timestamp, because the clients that need this most cannot. +* History retained on a consumer's behalf is bounded, and the bound holds while + the consumer is connected, not only across reconnections. +* The bound is expressed per consumer and does not require the consumer to + reason about the target's refresh or ingestion schedule. +* Retention is visible. An operator can see which consumer holds history and how + far behind it is. +* Losing the resume point is a named, loud failure. No configuration or failure + sequence may produce a silently skipped range of updates. +* The mechanism is reachable from the transports clients actually use: the + WebSocket SQL API, and pgwire from any mainstream driver. + +## Out of Scope + +* **Arbitrary SQL.** A subscription reads one collection, optionally through a + projection and filter. See "Query scope". +* **Indexes and views as targets.** An arrangement lives in cluster memory, so + no durable hold can be offered over it. +* **Exactly-once delivery.** See "Delivery semantics", which explains why this + is a deliberate step back from what the manual pattern can achieve, and how + such consumers are still served. +* **The PostgreSQL replication protocol.** `CopyBoth` is not implemented in + `src/pgwire/`, and "Alternatives" explains why adding it would not reach most + clients anyway. +* **Concurrent readers of one subscription.** A subscription is single-reader by + construction. +* **Changing materialized view self-correction.** This design consumes it and + does not modify it. +* **Fixing the as-of selection of newly created objects.** A subscription's hold + lowers the as-of chosen for objects created later over the same inputs. This + is a pre-existing property of as-of selection that the feature makes easier to + trigger. See "Known quirks and interactions". + +## Solution Proposal + +A durable subscription is a named catalog object that holds a read hold on a +storage collection, and that the consumer advances by acknowledging what it has +committed. The hold defines a window of readable time, the consumer's +acknowledgement is what moves the window's lower edge, and a wall-clock deadline +bounds how long the window may stay open without progress. A consumer that +reconnects reads from wherever inside that window it likes, and by default from +the position the server remembers. + +### Why a timestamp is a better resume token than a log position + +A Materialize timestamp is a global commit order, not a per-collection sequence +number, and that is the property that makes this feature worth more than a +resumable log tail. Two consequences matter. + +A single timestamp identifies a consistent cut across *every* collection in the +timeline, so a consumer reading several collections can assemble a transactional +snapshot by buffering each stream until its progress message passes `t`. Systems +built on write-ahead-log positions cannot do this from the positions alone, +because the positions of concurrent non-conflicting transactions interleave, so +they need transaction boundaries and synthetic pre-commit watermarks to recover +an order the timestamp gives for free. + +A timestamp is also meaningful to the client. It is comparable against +`mz_now()`, against other subscriptions, and against the timestamps the client +already sees in query results, whereas a log position is an opaque token whose +only defined operation is comparison with another token from the same source. + +### Read holds and read policies + +This design leans on a distinction the codebase makes in its mechanisms but has +never written down, and getting it wrong is what made the first draft of this +document unimplementable. The three concepts are: + +* A **read policy** is a rule that derives a frontier from a collection's write + frontier, re-evaluated as that frontier advances. `RETAIN HISTORY` and the + one-second default are policies. A policy expresses "keep this much history + behind the frontier", so it moves forward on its own. + +* A **read hold** is a token, acquired by a named party at a specific frontier + and released explicitly. `acquire_read_holds` grants one at the collection's + current `since` and returns `Result, CollectionMissing>`; + `ReadHold::try_downgrade` moves it and can fail. A hold expresses "this party + still needs to read from here". + +* The **capability**, and therefore `since`, is the meet of the policy-derived + frontier and every outstanding hold. + +A durable subscription is a party that holds a hold. It is emphatically not a +policy, and the first draft's `ReadPolicy::ValidFrom` plus +`ReadPolicy::Multiple` composition does not work: + +* `ReadPolicy::Multiple` has zero construction sites in the tree. It is an + unexercised variant, and `Multiple(vec![])` yields an empty antichain, which + drops the collection. +* Policy installation is a one-way ratchet. In + `src/storage-client/src/storage_collections.rs`, the new capability is applied + only `if PartialOrder::less_equal(&collection.implied_capability, &new_read_capability)`, + while `collection.read_policy = policy` is stored unconditionally. Installing + a lower frontier behind an already-advanced capability is a silent no-op, and + the trait documentation says so: it "will not 'recover' the read capability if + the prior capability is already ahead of it". +* The installation API cannot express it. Policies are derived from a bare + `CompactionWindow` at every call site, and bootstrap groups collections by + `CompactionWindow`, so there is no shape in which a per-collection constant + frontier survives. +* Any later `ALTER ... RETAIN HISTORY`, or the periodic metrics-retention + update, re-installs a window policy over the collection and would irreversibly + discard the subscription's contribution. +* Policies report nothing. `set_read_policies` returns `()`, and when persist + refuses a since downgrade the returned frontier is discarded by the caller. A + hold, by contrast, tells you the frontier you actually got. + +Holds also already interoperate with the rest of the system in the ways this +feature needs: `DROP` honors holds, and holds are deliberately acquired at the +earliest readable time specifically so that one controller can hold a frontier +back while another acquires at the same early point. + +### The readable window + +The subscription's durable state is a single frontier, `H`, the earliest as-of a +reader may request. The hold keeps `since <= H`, so the readable window is +`[H, upper)` and a reader may attach at any as-of inside it. + +`H` is derived from the acknowledgement, and the conversion is where the first +draft was wrong. The subscribe sink decides what to emit in +`src/compute/src/sink/subscribe.rs`: + +```rust +let beyond_as_of = if with_snapshot { + as_of.less_equal(time) +} else { + as_of.less_than(time) +}; +``` + +Without a snapshot, an as-of of `t` emits times **strictly greater** than `t`. +So a client that has processed everything before `t` and wants to resume at `t` +must read with as-of `t - 1`, which requires `since <= t - 1`. Storing the +acknowledged frontier and converting at every use site invites exactly the +off-by-one the manual pattern documents, so instead: + +**`ACKNOWLEDGE ... UP TO t` sets `H := t - 1`.** One conversion, in one place, at +the moment the claim is recorded. Everything downstream, the hold, the default +as-of, the window bound, and the introspection column, reads `H` directly and +performs no arithmetic. + +```text + H = ack - 1 + | + compacted |<------ readable window ------>| + ------------ [ ============================= ) -------> + upper +``` + +The two attach forms both land inside the window. `SNAPSHOT = false` with as-of +`H` yields times greater than `H`, meaning at or after the acknowledged +position, which is a gapless continuation. `SNAPSHOT = true` with as-of `H + 1` +yields a snapshot at the acknowledged position followed by later updates, which +is what a client that lost its local state needs. + +A client may also pass an explicit `AS OF` anywhere in the window, with the +ordinary meaning it has everywhere else in `SUBSCRIBE`. The window is what the +hold provides, and it is provided to whoever can read the collection: a plain +`SUBSCRIBE TO AS OF ` for any `x` in the window also works, and works +*because* a subscription is holding `since` back. + +### One hold, not two + +A running subscribe dataflow needs its input to stay readable, and today the +compute controller gives it a hold pinned at the dataflow's as-of that never +relaxes for the collection's lifetime. `forward_implied_capabilities` in +`src/compute-client/src/controller/instance.rs` is the only thing that would +relax it, and it returns early unless the cluster has no replicas, then skips +write-only collections with the comment "Collection is write-only, i.e. a sink." +Its own documentation names the consequence: forwarding is what "relaxes read +holds on inputs to forwarded collections, allowing their compaction". + +Left alone, that defeats the feature. A client that connects once and stays +connected acknowledges faithfully, its position tracks the write frontier, the +acknowledgement deadline never fires, and `since` sits at its attach-time as-of forever. +Retention would be bounded only across reconnections, which is the abnormal +case. + +Therefore the subscription's hold **is** the dataflow's input floor. The durable +attach installs no separate pinned hold; the dataflow reads behind the +subscription's hold, and the hold moves when the client acknowledges. This is +also why the general objection in `forward_implied_capabilities` does not apply +here. That comment worries that advancing a sink's capability could skip input +times across a replica restart, with no way to know whether the external +consumer had seen them. A durable subscription answers exactly that question: +`H` is the client's own statement about what it has processed, so restarting at +`H` is correct by construction rather than a guess. + +### Object model + +```mzsql +CREATE DURABLE SUBSCRIPTION ON WITH (ACKNOWLEDGE WITHIN ); +CREATE DURABLE SUBSCRIPTION WITH (ACKNOWLEDGE WITHIN ) AS ` form accepts a projection and filter over a single collection, +which is exactly the class that lowers to a map-filter-project pushed into the +persist read. That class needs no state, no dataflow, and no rehydration, and +therefore no reconciliation point of its own, which is what keeps it compatible +with a hold on the underlying collection. Filters and projections commute with +differencing, and a projection that collapses distinct rows merely consolidates +their diffs into a valid stream over the projected relation. + +Everything else is rejected: joins, aggregations, `DISTINCT`, subqueries, and +more than one collection in the `FROM` clause. Temporal filters over `mz_now()` +are rejected too, because they are not map-filter-project, they require a +dataflow, and they restructure retractions into the far future. + +The projection lives in the definition rather than being chosen per attach. Both +are sound, since `H` is a frontier on the underlying collection and any +projection over the same interval is individually correct, but definition-time +means the object fully describes its stream, introspection is meaningful, and a +client cannot change the row shape it deduplicates against between +reconnections. + +An error from the projection, such as a division by zero, surfaces as a stream +error. `H` does not move, because only `ACKNOWLEDGE` moves it, so the client +reconnects into the same error until the offending data changes. + +### Attaching + +```mzsql +SUBSCRIBE USING DURABLE SUBSCRIPTION WITH (PROGRESS, SNAPSHOT = false); +``` + +The form takes no target, because the object already names it. Omitting `AS OF` +resolves it from `H`, which is the intended path and the one that requires no +arithmetic from the client. + +`AS OF` is accepted, with the ordinary semantics it has everywhere else in +`SUBSCRIBE`, and must fall inside the window. It exists for consumers that track +their own position and are ahead of what they have acknowledged, which would +otherwise re-receive the whole unacknowledged window on every reconnection. The +alternative of forbidding it is recorded under "Alternatives". + +Keeping the ordinary semantics keeps the `-1`. Under `SNAPSHOT = false`, an +as-of of `t` emits times strictly greater than `t`, so a consumer resuming at its +committed position `T` must pass `T - 1`. This is a documented pitfall rather +than a solved problem, and it is a silent one: passing `T` skips every update at +`T`. Consumers that would rather not carry the arithmetic have a strictly safer +option, since updates carry `mz_timestamp`: omit `AS OF`, resume from `H`, and +discard everything below `T` on receipt. The failure mode of a filter is +redundant data; the failure mode of an off-by-one as-of is missing data. + +`UP TO` is also accepted. It constrains the upper edge, where the +`!up_to.less_equal(time)` boundary yields a clean half-open interval and no +arithmetic trap, so a batch consumer can drain up to a timestamp, commit, +acknowledge that same timestamp, and exit. + +Three clauses of plain `SUBSCRIBE` are rejected on this form. `ENVELOPE UPSERT` +and `ENVELOPE DEBEZIUM` cannot be produced correctly when resuming without a +snapshot, because neither the sink nor the server holds the prior value for a key +the resumed stream has not seen, and since `SNAPSHOT` defaults to `false` here +the broken combination would be the default one. `AS OF AT LEAST` asks the system +to choose a timestamp no earlier than a floor, which conflicts with resolving the +as-of from `H`. `WITHIN TIMESTAMP ORDER BY` is unaffected and remains available, +since it only orders rows within a timestamp. + +Rejecting the envelopes reads like ruling out the keyed consumer, so it is worth +saying why it does not. A consumer that maintains its own replica already holds +the prior value, which is why every sync engine surveyed keeps its own previous +state rather than asking the upstream for it. The raw diff stream is what those +consumers want, and it gives them more than a Postgres feed does: every +retraction carries the **full old row**, which is `REPLICA IDENTITY FULL` +semantics by construction, with no per-table configuration and no table +ownership. `WITHIN TIMESTAMP ORDER BY mz_diff` then orders every retraction +before every addition within a timestamp, which is exactly the ordering an +insert-or-replace store needs. So the keyed consumer is served by the default +form, not excluded from it. + +### Snapshot on resume + +**`SNAPSHOT` defaults to `false` on the durable form**, inverting the default of +plain `SUBSCRIBE`. Resuming is the whole point of the statement, and a default +that re-snapshots would make it useless without an explicit option. The +inconsistency is deliberate and worth the surprise. + +The snapshot, when requested, is always taken **at the acknowledged position**, +never at the latest time. A snapshot at the latest time would leave the client +holding state at a time it has not acknowledged and cannot name, silently +invalidating its position. + +That yields three modes, spanning what a resuming client can want: + +* **Continue.** `SNAPSHOT = false`, as-of `H`. The client holds state at or past + the acknowledged position and wants the diffs onward. The common case, and the + default. + +* **Reconcile at position.** `SNAPSHOT = true`, as-of `H + 1`. The client knows + its position but suspects its state has diverged, and wants the authoritative + state *at the position it claims* so it can compare. This is the same + operation the materialized view sink performs against its own output shard, + and it is readable because the hold keeps `since <= H`. + +* **Restart**, available only for an expired subscription: `WITH (RESET)`, or + `ALTER DURABLE SUBSCRIPTION ... RESET` then attach. The client + lost everything and wants current state. Reconciling at an old position would + deliver historical state plus every diff since, which is strictly more work + than a snapshot at the current time, and `RESET` already carries the right + meaning: consent to a gap. + +A live subscription deliberately cannot be fast-forwarded, by either spelling of +`RESET`, because that would discard data it is still holding history for and +would fence a reader that has not failed. A client that loses its local state +while its subscription is live therefore reconciles rather than restarting, and +pays a replay bounded by the deadline. That bound is what makes refusing the +fast-forward affordable. + +`WITH (PROGRESS)` is required whenever the client intends to acknowledge. A +progress message is the only thing that establishes that a timestamp is +complete: not every timestamp produces one, and a row at time `t` does not imply +that `t` is finished. Without progress messages a client cannot compute a safe +acknowledgement at all, so this is a requirement rather than a recommendation. + +**The attach must read the storage collection, not an index.** When an index +exists on the target, dataflow construction imports the index instead of the +source, which puts a compute collection in the id bundle, and subscribe +constrains its as-of by `least_valid_read()` over the whole bundle. The +subscription holds the storage `since`; the index's compute `since` follows the +one-second default, so a resume from an older position would fail with +`AdapterError::ImpossibleTimestampConstraints`. For an indexed relation, which +is the normal case for the console, that would make every resume fail. Reading +from storage is also the cheaper path, since it avoids rehydration. + +Attach takes an epoch and fences any previous reader, whose stream errors out. +Fencing rather than lease expiry suits the intended clients, because a +disconnected browser tab usually leaves a half-open connection the server has +not yet noticed, and waiting for a lease to expire would make a legitimate +reconnect fail for seconds. + +### Acknowledging + +`ACKNOWLEDGE DURABLE SUBSCRIPTION UP TO ` asserts that the client +has durably processed every update at times strictly before the given timestamp, +and sets `H := timestamp - 1`. + +**`ACKNOWLEDGE` must not be classified as DDL.** `must_serialize_ddl` returns +early when `!StatementClassification::from(stmt).is_ddl()`, and a DDL +classification would be actively harmful: the first acknowledgement in a +subscribe transaction would take the environment-wide `serialized_ddl` lock, +held until the transaction ends and therefore for the life of the subscribe, +blocking all other DDL; a second one in the same transaction would soft-panic. +The statement needs its own non-DDL durable write path. + +`ACKNOWLEDGE` is monotone, idempotent, and non-transactional. It takes effect +immediately and is not rolled back, because the client really did commit the +data. A value at or below the current position is accepted and has no effect +rather than being rejected, so that a client retrying after an ambiguous failure +does not have to distinguish "already applied" from "too low". A value beyond +the target's write frontier is an error, since the client cannot have processed +what was never sent. Ordering against the row stream is irrelevant, since an +acknowledgement is a claim about the client's own state, which is also what +allows the statement to arrive on any connection. + +The bound is exclusive, matching `UP TO` on `SUBSCRIBE` rather than introducing a +third convention. A progress message at `t` states that nothing further will +arrive strictly before `t`, and `ACKNOWLEDGE ... UP TO t` states that everything +strictly before `t` is processed, so the number a client reads from the progress +message is the number it sends back unchanged. Reading `UP TO t` and then +acknowledging `UP TO t` is likewise symmetric. + +The in-memory value is a coalescing buffer. A periodic task writes it durably +and downgrades the hold. Retained history is therefore the client's true +position plus at most one flush interval. Because the durable write is what +authorizes compaction, a lost flush costs retention, never correctness. + +Cursor state lives in a dedicated durable `StateUpdateKind` rather than in +`create_sql`. Rewriting `create_sql` per flush is possible, and +`Op::AlterRetainHistory` is precedent for mutating a durable item that way, but +it would make every flush a full catalog transaction with an audit-log entry and +turn `SHOW CREATE` into a moving value. `create_sql` is a definition and `H` is +state. + +At the scale this must support, the flush is the hot path. See "Scale". + +### Expiry + +The acknowledgement deadline is measured on the **wall clock**, as time since the last +acknowledgement, not as a distance between `H` and the write frontier. + +Anchoring it to the write frontier fails in both directions. A `REFRESH` MV +rounds its frontiers up to the next refresh, so its lag is at least one refresh +interval even for a client that acknowledges instantly, and a correct deadline +would have to be aware of every refresh policy in the object's ancestry, +the way replica expiration computes its offset. In the other direction a stalled +source, a paused source, or a zero-replica cluster freezes the frontier, so a +frontier-anchored acknowledgement deadline never fires precisely when releasing the hold +matters most. A wall clock is uniform across every target type and needs no +knowledge of the target's schedule. + +The last-acknowledgement wall-clock time is recorded durably alongside `H`, +which the periodic flush is already writing. Keeping it only in memory would +grant every subscription a fresh deadline on every environment restart. + +The ceiling on the deadline should be expressed in **hours**, not minutes. A +minute suits an interactive client, but a server-side consumer's outages are +deploys, crash loops, and image rollbacks, which are tens-of-minutes events, and +a ceiling below that turns every such event into a full resync. Relatedly, a long +*environment* outage should pause the deadline rather than consume it: the +subscription's client was not given a chance to acknowledge, so expiring it on +boot would punish it for our downtime. + +On expiry the subscription is marked expired durably and only then is the hold +released. Ordering matters: releasing first and crashing before recording expiry +would leave a subscription that looks valid while `since` has advanced past its +`H`. Recording expiry first is self-healing, because an expired subscription +acquires no hold on boot. The reverse crash, a recorded expiry with the hold +still held, resolves itself the same way. + +The object survives expiry, which is what makes the state reportable. Attach +fails with an error naming the subscription, the position it expired at, and the +lag that killed it. `ALTER DURABLE SUBSCRIPTION RESET` re-arms it at the +current frontier, preserving identity, ownership, and grants, and making the +client's consent to a gap explicit. `RESET` on a subscription that has not +expired is an error, since it would silently fence a live reader. + +Consent may also be given **in band**, so that a server-side consumer recovering +from an outage needs no privileged data-definition statement in its reconnect +path. Every sync engine surveyed has such a signal, under names like +`reset-required`, `must-refetch`, and `CLEAR`. The mechanism is two attaches +rather than one option: + +* A plain attach on an expired subscription **fails**, with an error + distinguishable from every other attach failure so a client can branch on it. +* `SUBSCRIBE ... WITH (RESET)` resets the subscription to the current time and + delivers a snapshot there. It errors if the subscription has *not* expired, so + it cannot silently fence a live reader. + +Two statements rather than one, because of how a consumer knows it received a +fresh snapshot rather than a continuation. A single reset-if-needed attach cannot +tell it, and the stream has no channel for the answer: a snapshot is a batch of +`+1` diffs at the as-of, indistinguishable from genuine insertions at that +timestamp, and signalling it otherwise would mean either a new column on every +`SUBSCRIBE` or a session notice, which is advisory and routinely ignored. With +two attaches the client knows because it *asked*, having branched on the expiry +error. One extra round trip, only on the rare expiry path, for an unambiguous +answer. + +The opening progress message is a useful cross-check but not a substitute for +this. `SUBSCRIBE`'s first emitted update is guaranteed to be a progress message +carrying the as-of, so a client that retained its own last acknowledged position +can compare the two and detect a gap. But a client that relies on the server to +hold its position is exactly the one that may not have retained it, so the +guarantee cannot carry the signal on its own. + +`ALTER ... RESET` remains for operators. + +```mermaid +stateDiagram-v2 + [*] --> Active: CREATE DURABLE SUBSCRIPTION + Active --> Active: ACKNOWLEDGE + Active --> Expired: no ACKNOWLEDGE within the deadline + Expired --> Active: ALTER ... RESET + Active --> [*]: DROP + Expired --> [*]: DROP +``` + +Expiry is evaluated by the same periodic task that flushes, which already holds +the state it needs. Evaluating it lazily at attach would need no background task +but would never release an abandoned subscription's hold. + +`CREATE` must reject a deadline below a fixed multiple of the flush cadence. The +value compared against the deadline is the durable one, which trails the client's +claim by up to one flush interval, so a deadline near the cadence would expire +clients that are acknowledging correctly. + +### Delivery semantics + +A durable subscription is **at-least-once**. The client commits, then +acknowledges, and a failure in between means re-delivery. + +The idempotence rule is per timestamp, not per row: **apply each timestamp +atomically, record which timestamp you applied, and skip timestamps already +applied.** Deduplicating on the timestamp and row is not sufficient, because a +resumed interval may consolidate differently than it did originally, so the same +logical change can arrive as a different set of `(row, diff)` tuples. It is also +not necessary, because resume always lands on a timestamp boundary, so +re-delivery is always of whole timestamps. Every sync engine surveyed already +works this way, applying one upstream commit atomically and recording the +resulting version inside the same transaction. + +A consumer must therefore not treat the stream as an event log. + +This is a deliberate step back from what the manual pattern achieves, and the +existing user documentation is explicit that writing the data and the position +in one transaction is what makes that pattern exactly-once. A server-side +cursor structurally cannot offer that, because the acknowledgement is a separate +round trip after the commit. + +Such consumers are still served. One that records its own committed position `T` +in the same transaction as the data keeps exactly-once, and the subscription +supplies the retention guarantee that the history back to `H` is still there. It +has two ways to skip what it already has: + +* **Discard on receipt.** Resume from `H` and drop every update below `T`. Costs + bandwidth proportional to the unacknowledged window, and cannot lose data. +* **Position the read.** Pass `AS OF T - 1`. Costs nothing, and loses data + silently if the arithmetic is wrong. + +Filtering is the better default and positioning is the better optimization. The +consumer keeps owning correctness either way, which is where it has to live, +since only the consumer knows what it committed. + +### Scale + +The target is one to ten thousand concurrent subscriptions, which are +**provisioned per logical consumer** and reused across reconnections, not +created per session or per page load. The distinction is load-bearing: creating +a subscription is a catalog transaction on the coordinator, and every durable +item is replanned at boot, so churn is far more expensive than population. + +Three consequences for the implementation: + +* **The flush must batch.** Ten thousand subscriptions flushing individually + once a second is ten thousand catalog transactions per second, which is not + viable. One durable write must carry many cursors, and the cadence must scale + with population rather than being a fixed per-subscription interval. + +* **The per-collection minimum must be incremental.** Recomputing a + collection's floor by scanning its subscriptions on every flush is quadratic + in the shared-collection case, which is the case ten thousand implies. + +* **Boot cost is linear in population.** Ten thousand items add replanning work + at startup. This is the price of the catalog-item model and the reason + subscriptions must not be per-session. + +### Observability + +`mz_durable_subscriptions` reports each subscription's name, target, owner, time +to live, `H`, wall-clock time since last acknowledgement, validity state, and +the position at which an expired subscription died. The cursor rows are +available through `mz_catalog_raw`, which exposes durable `StateUpdateKind` +rows, but the lag column is not catalog state: write frontiers live in +`mz_internal.mz_frontiers`, so the relation needs a join against an +introspection source. `mz_catalog_raw` is system-user only, so the +operator-facing relation needs its own grants. + +The name is deliberately distinct from `mz_internal.mz_subscriptions`, which +lists running `SUBSCRIBE` statements and is what the console and the in-tree +cancellation tests join against. A durable subscription appears in the new +relation whether or not anything is reading it. `mz_history_retention_strategies` +is the closest prior art for retention observability and the new relation should +be consistent with it. + +A `parse_catalog_create_sql` arm is required. Contrary to a claim in an earlier +draft, that function has no catch-all: the match is exhaustive by design, with +the in-source rationale that "one unclassified `create_sql` takes out +`mz_objects`, `mz_indexes`, and every sibling view at once", so a new statement +variant is a compile error. The real hazard is landing in the reject group where +`Subscribe(_)` already sits, which would break every catalog view that scans +`Item` rows. + +### Protocol + +Which acknowledgement channel is available depends on how the client reads, and +the binding constraint is a property of the PostgreSQL protocol rather than of +any driver. During plain `COPY OUT` there is no legal frontend-to-server data +path: the protocol documentation states the frontend cannot abort the transfer +except by closing the connection or issuing a cancel request, and libpq's +`PQputCopyData` admits `COPY_IN` and `COPY_BOTH` while excluding `COPY_OUT`. +Drivers then fail in incompatible ways, from silent indefinite queueing in +node-postgres to a thread block in pgjdbc to an immediate error in pgx, asyncpg, +and postgres.js to a compile error in sqlx. Our own loop matches the constraint: +it selects on `wait_closed`, which polls socket readiness on a timer and never +reads bytes. + +| Read path | Ack channel | +| --- | --- | +| `DECLARE` plus `FETCH` | `ACKNOWLEDGE` between `FETCH`es, same connection | +| Streaming `COPY OUT` | Second connection, protocol-mandated | +| WebSocket | Same connection, inbound frame | + +Interleaving between `FETCH`es is reachable from every mainstream driver, +because each `FETCH` is a self-contained round trip that releases whatever +per-connection lock the driver holds, and it is already what our documentation +recommends. Naming the object is what makes the second-connection path clean: +cancellation must locate a session through `mz_subscriptions` and `mz_sessions`, +whereas an acknowledgement addresses the subscription by name. + +An earlier draft claimed all three paths need `TransactionOps::Subscribe` +relaxed. They probably do not. That op is recorded only when the subscribe's +`when` is `QueryWhen::Immediately`, and a durable attach resolves an as-of, so +the durable form does not enter the gate at all. The load-bearing protocol +requirement is the non-DDL classification described under "Acknowledging". + +For WebSocket the server is half-duplex per statement: `run_ws` reads one +request, executes it to completion while holding the socket, and only then reads +again, while `connection_error` writes a periodic ping without ever reading. The +minimal change is a tight allowlist rather than general concurrency: split the +socket, add one inbound arm to the subscribe loop's `select!`, and process only +`ACKNOWLEDGE` there. That arm is polled in a loop so it must be cancel-safe, and +`connection_error` doubles as the liveness prober, which wants revisiting once a +real reader exists. + +### Testing + +Testdrive covers the lifecycle: create, attach, acknowledge, detach, reattach, +and assert the resumed stream contains neither a snapshot nor a gap. The gap +assertion is the one that matters, and the `H = ack - 1` boundary is what it +tests, so it must include an update at exactly the acknowledged timestamp. + +A restart test is mandatory, and its purpose is to prove that the hold is +re-acquired from durable state where a policy would silently ratchet away. Note +that an earlier draft justified a restart platform check as catching an +`enable_for_item_parsing` boot failure, which is self-refuting: CI runs new +feature flags on, so such a check would run with the flag enabled and pass. The +boot-failure risk is real but is a release-ordering concern, not a testable one. + +Then the cases that would otherwise fail silently: + +* **An index on the target.** Attach must still resume, since dataflow + construction would otherwise import the index and constrain the as-of by its + compute `since`. Include this in the earliest test, because it is the normal + case for an indexed relation and would break every resume. +* **Fencing**, with two connections, asserting the first stream errors. +* **Expiry**, asserting the error names the subscription. The deadline is + wall-clock, so drive it with a very short deadline rather than by waiting. +* **`DROP` of the target without `CASCADE`** fails while a subscription exists. +* **Rejected surface**: temporal filters, multi-collection `FROM`, joins and + aggregations, `ENVELOPE UPSERT`, `ENVELOPE DEBEZIUM`, and `AS OF AT LEAST`. +* **Acknowledging at or below the current position** is accepted as a no-op. +* **A consistent cut across two subscriptions**: buffer both to the same + timestamp and assert the union matches a `SELECT ... AS OF` at that timestamp. + This is the property that makes the single-target restriction tolerable, so it + should be asserted rather than assumed. + +### Known quirks and interactions + +These are consequences of existing behavior that this design does not fix. +Documenting them is deliberate; fixing them is separable work. + +* **A subscription's hold lowers the as-of of objects created later.** + `CREATE MATERIALIZED VIEW`, `CREATE INDEX`, and `CREATE SINK` all take their + as-of from `least_valid_read()` over their inputs, so one lagging subscriber + makes a subsequent create backfill the whole retained history. For + materialized views that as-of is written durably into `create_sql`. Retention + is therefore not consumer-local, and the wall-clock deadline is the only thing + bounding the blast radius. + +* **Read-only mode and zero-downtime upgrade.** Every existing read policy is a + function of the write frontier, which is what makes two generations agree + without coordination; `H` is durable state, so it is the first floor two + generations could disagree about. Concretely: `SUBSCRIBE` is permitted in + read-only mode while a new `Plan::Acknowledge` would fall into the planner's + catch-all and be refused, so clients would attach, stream, and have every + acknowledgement rejected until the deadline killed a subscription whose client + did nothing wrong. The flush and expiry task must be read-only gated, + like every comparable periodic coordinator task. Note that the obvious gate is + the wrong one: savepoint mode, which zero-downtime upgrade uses, reports + itself as not read-only. + +* **Name transforms and hand-maintained matches.** `src/sql/src/ast/transform.rs` + carries several matches over statement kinds that end in `unreachable!()`, and + the most recently added item kind is missing from one of them. These run for + every item in a renamed schema, so `ALTER SCHEMA ... SWAP`, the documented + blue/green cutover, panics the coordinator without a new arm. + +* **Blue/green cutover is a name swap.** Ids and shards are untouched, so a + subscription keeps streaming the decommissioned collection, and the + documented teardown then produces exactly the "does not exist" ambiguity this + design rejects for expiry. + +* **`ALTER TABLE ... ADD COLUMN`** is gated behind `enable_alter_table_add_column`, + which defaults to false and is undocumented, so this is a forward-looking note + rather than a behavior users can reach. If it is enabled, the statement mints a + new `GlobalId` on the same shard while the hold survives via primary links, and + a subscription on the target would see its row arity change with no marker + saying why. Expiring the subscription is the conservative answer. A better one, + should schema change become supported, is to keep the hold and `H` and fail + only the in-flight attach, so a reconnect resumes at the same position with the + new shape: the arity changes *at a timestamp*, so a boundary does exist even + under `SNAPSHOT = false`. + +* **Combining several subscriptions is safe but unassisted.** Because a timestamp + is a global cut, a consumer reading N collections can buffer each stream until + its progress message passes `t` and then apply the union at `t` as a + transactionally consistent snapshot. That works today and needs nothing from + us, but the consumer pays for it: liveness is the minimum over N, so one + `REFRESH` target paces the whole set; expiry is per-subscription, so one + expiry breaks the joint cut and recovery is per-collection; and each checkpoint + costs N `ACKNOWLEDGE` statements. The recipe should be documented, and it is + the argument for the multi-target follow-up. + +* **Timelines are not comparable.** Collections in different timelines have + incomparable timestamps, so the recipe above is unsound across them. A + subscription should reject a target outside the default `EpochMilliseconds` + timeline, or the restriction must be documented. + +* **There is no mapping from an upstream position to a Materialize timestamp.** + A consumer that writes to an upstream database and then needs to know when its + own write became visible in Materialize cannot ask. This is invisible if only + the read path is modeled, and it is what a sync engine needs to retire an + optimistic write. Out of scope here, but it is a hard dependency for that class + of consumer and is worth naming so nobody assumes this feature covers it. + +* **The WebSocket transport has no batch flow control.** That API supports + neither `DECLARE`, `FETCH`, `CLOSE`, nor `COPY`, so a WebSocket reader cannot + pull in batches and depends entirely on the inbound-`ACKNOWLEDGE` arm described + under "Protocol". For the console, which is the intended first consumer, that + arm is not an optimization but the only acknowledgement channel. + +* **Rollback ordering.** `item_type()` panics on an unknown `create_sql` prefix, + and topological sorting parses every item's `create_sql` with `expect`. Both + sit upstream of the feature flag, so once a single row exists, rolling back to + a binary that predates the parser makes the environment unbootable. This is a + release-ordering constraint, not something a test catches. + +* **Empty and regressing frontiers.** The tree already carries three + incompatible conventions for the lag of a collection with an empty upper. The + wall-clock deadline sidesteps this for expiry, but the introspection lag + column still has to pick one. + +* **Stale as-ofs on transaction-managed tables** are supported by design but are + not a hot path today, and read-only mode has no transaction-shard handle at + all. Which of the logical and physical uppers acknowledgement validation + compares against needs to be pinned down. + +## Minimal Viable Prototype + +The prototype is the hold path end to end against a table, with no expiry, no +fencing, no projection, and no observability. Create the object, acquire a hold +and downgrade it from a synchronously written `H`, implement `ACKNOWLEDGE` as a +non-DDL statement, and drive it from testdrive: acknowledge, restart +`environmentd`, reattach with `SNAPSHOT = false`, and assert the stream resumes +with neither a snapshot nor a gap. + +That validates the four claims this design rests on and is cheapest to be wrong +about: that a hold re-acquired from durable state genuinely survives a restart +where a policy would not, that `H = ack - 1` puts the boundary in the right +place, that a non-DDL `ACKNOWLEDGE` neither deadlocks nor panics inside a +subscribe transaction, and that the attach reads storage rather than an index. +Add an index to the target as part of the test, since that is the case that +would otherwise fail every resume. + +It deliberately does not validate the batched flush, the wall-clock deadline, or +the WebSocket path. + +A second spike is worth running against the console, the intended first +consumer, whose reconnect logic this feature replaces: point `SubscribeManager` +at a durable subscription and delete the `resubscribing` path. + +## Alternatives + +**A read policy contribution rather than a hold.** This was the first draft, and +"Read holds and read policies" records why it fails: `ReadPolicy::Multiple` is +unconstructed, installation only ratchets capabilities upward, the installation +API is keyed by `CompactionWindow`, an ordinary `ALTER ... RETAIN HISTORY` would +discard the contribution, and nothing in the path reports failure. + +**Resuming from `since` with no durable position.** Since compaction is monotone +and durable in persist, `since` is itself a record of progress, suggesting a +resume at `max(H, since)` with nothing stored hot. The default compaction window +is one second, so `since` records what other readers permit rather than what +this consumer consumed, and it is exact only when the subscription is the sole +reader. Worse, when the in-memory position has moved past the last flush, a +restart would resume above the durable position and skip the difference. Safe +under the commit-before-acknowledge contract, but silent. + +**A duration-based floor, meaning `RETAIN HISTORY` alone.** This is the shipped +manual pattern. A duration is a guess in both directions, too short bricking the +client with no explanation and too long growing storage with no consumer to +attribute it to, and it leaves the client responsible for remembering a +timestamp. It remains the right answer for exactly-once consumers, which is why +it stays documented rather than being replaced. + +**A frontier-anchored acknowledgement deadline.** More directly expresses "bound the +retained history", and was the first draft's choice. It requires every consumer +of a `REFRESH` MV to reason about every refresh policy in the ancestry, and it +never fires for a frozen frontier, which is when releasing the hold matters +most. + +**An object owning its own query and output shard.** Self-contained, with one +`DROP` cleaning up compute, storage, and cursor together, and an unambiguous +hold because the reader is the only consumer. It duplicates compute and storage +whenever consumers share a query, and one object per consumer does not reach the +target scale. + +**Cursor state in `create_sql`, or outside the catalog.** The former makes every +flush a catalog transaction with an audit entry and a moving `SHOW CREATE`; the +latter is cheapest per write but reintroduces two durable locations that must +agree on create and drop. + +**Forbidding `AS OF` on the durable form.** Considered, because the `less_than` +boundary makes a client-supplied timestamp a silent-data-loss pitfall, and +because discarding on `mz_timestamp` covers the same need with a failure mode of +redundancy rather than silence. Rejected because avoiding one documented pitfall +is the only thing it buys, while the cost is real: a consumer that acknowledges +lazily would re-receive its entire unacknowledged window on every reconnection, +with no way to opt out. Two positioning conventions, an inclusive keyword for the +durable form alongside the exclusive `AS OF` elsewhere, was also considered and +rejected as worse than the pitfall, since it makes users learn a second +convention that exists in exactly one place. + +**Non-exclusive readers.** Taking the maximum of concurrent acknowledgements +needs no epoch, but two readers would advance each other's floor past unread +data, which is undetectable data loss. + +**Advancing the hold on expiry instead of expiring the subscription.** Reclaims +storage while keeping the subscription usable, at the cost of the client +receiving a later stream and never learning a gap occurred. + +**Dropping the object on expiry.** Frees the name, but "does not exist" is +indistinguishable from a typo, so the client cannot tell it must handle a gap. + +**`CopyBoth`.** The idiomatic PostgreSQL answer, with ordering for free. It does +not exist in `src/pgwire/`, we are only a `CopyBoth` client against upstream +PostgreSQL, it is absent from several major drivers and ack-only in others, and +it inherits the walsender simple-query restriction and pooler refusal. + +## Follow-up work + +**A consistent cut across several collections.** One subscription over multiple +collections sharing a position would deliver a resumable, consistent +multi-relation cut, which is the capability that replication slots, warehouse +change streams, and log-based consumer groups do not provide. The single-target +design here is the smaller first step, and multiple targets extend the same hold +and cursor machinery rather than requiring a different one. + +**As-of selection for newly created objects.** The quirk above is worth fixing +on its own merits: a new dataflow needs a readable as-of, not the oldest +readable one. + +**Relaxing sink input holds in general.** This design special-cases durable +subscriptions because their consumer's position is known. The same reasoning may +generalize to other sinks that track consumer progress. + +## Open questions + +* No tracking issue exists yet. One must be filed and linked above before this + document merges. +* What are the default flush cadence, the minimum deadline as a multiple of + it, and the dyncfg ceiling on the deadline? The ceiling should be in hours per + "Expiry", but the value is undecided. +* Can the durable attach be made to bypass index import cleanly, or does that + need a change to dataflow construction? This is the one implementation + question that could change the shape of the attach path. +* How is the wall-clock acknowledgement deadline evaluated across a long environment + outage? Recording the last-acknowledgement time durably means a multi-hour + restart expires every subscription on boot, which is defensible but should be + a deliberate choice. +* Is `ALTER DURABLE SUBSCRIPTION ... RESET` in the first release, or is `DROP` + and `CREATE` acceptable initially at the cost of re-granting privileges? +* Can retained bytes be attributed to an individual subscription, or only to the + collection? +* Which upper does acknowledgement validation compare against for a + transaction-managed table, logical or physical? +* Once the WebSocket subscribe loop has a real reader, should liveness move from + writing pings to observing pongs? diff --git a/doc/user/content/headless/sql-command-privileges/acknowledge.md b/doc/user/content/headless/sql-command-privileges/acknowledge.md new file mode 100644 index 0000000000000..bbd625bb9ffc5 --- /dev/null +++ b/doc/user/content/headless/sql-command-privileges/acknowledge.md @@ -0,0 +1,5 @@ +--- +headless: true +--- +- Ownership of the durable subscription. +- `SELECT` privileges on the object it subscribes to. diff --git a/doc/user/content/headless/sql-command-privileges/alter-durable-subscription.md b/doc/user/content/headless/sql-command-privileges/alter-durable-subscription.md new file mode 100644 index 0000000000000..35c8b37a0f09a --- /dev/null +++ b/doc/user/content/headless/sql-command-privileges/alter-durable-subscription.md @@ -0,0 +1,4 @@ +--- +headless: true +--- +- Ownership of the durable subscription. diff --git a/doc/user/content/headless/sql-command-privileges/create-durable-subscription.md b/doc/user/content/headless/sql-command-privileges/create-durable-subscription.md new file mode 100644 index 0000000000000..eb7b0ec6fedda --- /dev/null +++ b/doc/user/content/headless/sql-command-privileges/create-durable-subscription.md @@ -0,0 +1,6 @@ +--- +headless: true +--- +- `CREATE` privileges on the containing schema. +- `SELECT` privileges on the object being subscribed to. +- `USAGE` privileges on the schema containing that object. diff --git a/doc/user/content/headless/sql-command-privileges/drop-durable-subscription.md b/doc/user/content/headless/sql-command-privileges/drop-durable-subscription.md new file mode 100644 index 0000000000000..35c8b37a0f09a --- /dev/null +++ b/doc/user/content/headless/sql-command-privileges/drop-durable-subscription.md @@ -0,0 +1,4 @@ +--- +headless: true +--- +- Ownership of the durable subscription. diff --git a/doc/user/content/sql/acknowledge.md b/doc/user/content/sql/acknowledge.md new file mode 100644 index 0000000000000..77e4892fca47f --- /dev/null +++ b/doc/user/content/sql/acknowledge.md @@ -0,0 +1,131 @@ +--- +title: "ACKNOWLEDGE" +description: "`ACKNOWLEDGE` advances the position of a durable subscription." +menu: + main: + parent: 'commands' +--- + +{{< private-preview />}} + +`ACKNOWLEDGE` tells Materialize how far you have processed a [durable +subscription](/sql/create-durable-subscription/), which advances the position it +resumes from and releases the history before it. + +## Syntax + +```mzsql +ACKNOWLEDGE DURABLE SUBSCRIPTION UP TO +; +``` + +| Field | Use | +| --- | --- | +| `` | The durable subscription to advance. | +| `` | An [`mz_timestamp`](/sql/types/mz_timestamp/). Asserts that you have durably processed every update at times **strictly before** this value. | + +## Details + +### What to acknowledge + +Acknowledge the `mz_timestamp` of a progress message, which is why +[`SUBSCRIBE`](/sql/subscribe/#progress) must be run `WITH (PROGRESS)` when you +intend to acknowledge. A progress message with timestamp `t` means no further +updates will arrive at times strictly before `t`, which is exactly the claim +`ACKNOWLEDGE ... UP TO t` makes back to Materialize. The bound is exclusive in +both directions, so the number you read from the progress message is the number +you send back unchanged. + +Do not acknowledge the timestamp of an ordinary row. Not every timestamp +produces a progress message, and a row at time `t` does not mean that time `t` +is complete, so acknowledging it can skip updates you have not seen. + +`UP TO` is exclusive here for the same reason it is exclusive on +[`SUBSCRIBE`](/sql/subscribe/#up-to), which makes the batch pattern symmetric: +read `UP TO` a timestamp, then acknowledge `UP TO` that same timestamp. + +### Order of operations + +Commit your data first, then acknowledge. If you acknowledge before your own +processing is durable, and your application then fails, the acknowledged updates +are gone and cannot be re-delivered. + +### Semantics + +`ACKNOWLEDGE` is: + +* **Monotone.** Acknowledging a timestamp at or below the current position is + accepted and has no effect, so retrying is safe. + +* **Idempotent.** Sending the same acknowledgement twice is indistinguishable + from sending it once. + +* **Not transactional.** The acknowledgement takes effect immediately and is + not undone by `ROLLBACK`. This is deliberate: your data really was + committed, so rolling back must not un-acknowledge it. + +Acknowledging a timestamp beyond the object's write frontier is an error. The +frontier is the largest timestamp for which the subscription could have sent you +a progress message, and it is reported for every object in +[`mz_internal.mz_frontiers`](/reference/system-catalog/mz_internal/#mz_frontiers). + +### Where you can run it + +`ACKNOWLEDGE` may be run on the same connection as the subscription, interleaved +between `FETCH` statements, or on a separate connection. Because a subscription +is named, no coordination between connections is needed, and because +acknowledgements are monotone, they cannot arrive out of order in any way that +matters. + +Running `ACKNOWLEDGE` while no one is reading the subscription is allowed. This +matters for the separate-connection case, where the reading connection may drop +while an acknowledgement is in flight. + +### Effect on resuming and on storage + +The acknowledged position is where the subscription resumes, and it determines +how much history Materialize retains. Acknowledging more often releases storage +sooner and shortens the replay after a failure; acknowledging less often reduces +round trips. Materialize records the position durably on a short interval rather +than on every statement, so history is released slightly after you acknowledge. + +When you resume without an explicit `AS OF`, Materialize positions the +subscription so that you receive updates at and after the acknowledged time. With +`SNAPSHOT true` you receive the state *as of* that time, with updates at that +time already folded in, followed by later updates. Either way you do not subtract +anything. The subtraction is only needed if you choose to pass [`AS +OF`](/sql/create-durable-subscription/#where-reading-starts-and-stops) yourself, +which is an exclusive bound. + +## Examples + +Acknowledging from a progress message received on the same connection: + +```mzsql +FETCH ALL c WITH (timeout = '1s'); +``` + +```nofmt + mz_timestamp | mz_progressed | mz_diff | auction_id | amount +---------------+---------------+---------+------------+-------- + 1723459199000 | f | 1 | 1 | 42 + 1723459200000 | t | | | +``` + +```mzsql +ACKNOWLEDGE DURABLE SUBSCRIPTION winning_bids_feed UP TO 1723459200000; +``` + +## Privileges + +The privileges required to execute this statement are: + +{{% include-headless "/headless/sql-command-privileges/acknowledge" %}} + +## Related pages + +* [`CREATE DURABLE SUBSCRIPTION`](/sql/create-durable-subscription/) +* [`ALTER DURABLE SUBSCRIPTION`](/sql/alter-durable-subscription/) +* [`DROP DURABLE SUBSCRIPTION`](/sql/drop-durable-subscription/) +* [`SUBSCRIBE`](/sql/subscribe/) +* [Resuming subscriptions](/transform-data/patterns/durable-subscriptions/) diff --git a/doc/user/content/sql/alter-durable-subscription.md b/doc/user/content/sql/alter-durable-subscription.md new file mode 100644 index 0000000000000..14a44f6d0487f --- /dev/null +++ b/doc/user/content/sql/alter-durable-subscription.md @@ -0,0 +1,85 @@ +--- +title: "ALTER DURABLE SUBSCRIPTION" +description: "`ALTER DURABLE SUBSCRIPTION` changes the acknowledgement deadline of a durable subscription, resets one that has expired, or transfers ownership." +menu: + main: + parent: 'commands' +--- + +{{< private-preview />}} + +`ALTER DURABLE SUBSCRIPTION` changes the acknowledgement deadline of a [durable +subscription](/sql/create-durable-subscription/), resets one that has expired, or +transfers its ownership. + +## Syntax + +```mzsql +ALTER DURABLE SUBSCRIPTION SET (ACKNOWLEDGE WITHIN ); +ALTER DURABLE SUBSCRIPTION RESET; +ALTER DURABLE SUBSCRIPTION OWNER TO ; +``` + +| Field | Use | +| --- | --- | +| `SET (ACKNOWLEDGE WITHIN )` | Change how long you may go without acknowledging. Takes effect immediately, including for a subscription that is currently behind. | +| `RESET` | Re-arm an expired subscription at the current time. | +| `OWNER TO ` | Transfer ownership to another role. | + +## Details + +### Changing the acknowledgement deadline + +Increasing the deadline gives a reader more time to recover, and increases the +history that may be retained on its behalf. Decreasing it can expire a +subscription immediately, if the reader has already gone longer than the new +value without acknowledging. + +The value must fall between the system-wide minimum and maximum, the same bounds +[`CREATE DURABLE SUBSCRIPTION`](/sql/create-durable-subscription/#acknowledgement-deadline) +enforces. A minimum exists because the acknowledged position is recorded durably +on an interval, so a deadline close to that interval would expire readers that +are acknowledging correctly. + +### Resetting an expired subscription + +`RESET` moves an expired subscription to the current time and makes it usable +again. It does not recover the history that was released when the subscription +expired, so the next read must request `SNAPSHOT true` to get a usable starting +state. + +Use `RESET` rather than dropping and recreating: it preserves the subscription's +name, owner, and privileges. Requiring it, instead of silently resuming from +whatever history happens to remain, is what keeps a gap in the data from passing +unnoticed. + +`RESET` on a subscription that has not expired is an error. Fencing a live reader +by resetting its position is not something to do by accident. + +## Examples + +```mzsql +ALTER DURABLE SUBSCRIPTION winning_bids_feed SET (ACKNOWLEDGE WITHIN '5m'); +``` + +```mzsql +ALTER DURABLE SUBSCRIPTION winning_bids_feed RESET; +``` + +```mzsql +ALTER DURABLE SUBSCRIPTION winning_bids_feed OWNER TO analytics_owner; +``` + +## Privileges + +The privileges required to execute this statement are: + +{{% include-headless "/headless/sql-command-privileges/alter-durable-subscription" %}} + +## Related pages + +* [`CREATE DURABLE SUBSCRIPTION`](/sql/create-durable-subscription/) +* [`ACKNOWLEDGE`](/sql/acknowledge/) +* [`DROP DURABLE SUBSCRIPTION`](/sql/drop-durable-subscription/) +* [`SUBSCRIBE`](/sql/subscribe/) +* [Resuming subscriptions](/transform-data/patterns/durable-subscriptions/) diff --git a/doc/user/content/sql/create-durable-subscription.md b/doc/user/content/sql/create-durable-subscription.md new file mode 100644 index 0000000000000..2eb3c1e3ee6f5 --- /dev/null +++ b/doc/user/content/sql/create-durable-subscription.md @@ -0,0 +1,441 @@ +--- +title: "CREATE DURABLE SUBSCRIPTION" +description: "`CREATE DURABLE SUBSCRIPTION` creates a named, resumable subscription whose progress Materialize tracks for you." +menu: + main: + parent: 'commands' +--- + +{{< private-preview />}} + +`CREATE DURABLE SUBSCRIPTION` creates a named subscription that Materialize can +resume. Materialize tracks how far you have processed, and retains exactly the +history you still need. + +## Conceptual framework + +A plain [`SUBSCRIBE`](/sql/subscribe/) has no memory. If the connection drops, +the next `SUBSCRIBE` starts over, and unless you configured a [history retention +period](/transform-data/patterns/durable-subscriptions/) and recorded your own +timestamps, it starts over with a full snapshot. + +A durable subscription moves both of those responsibilities into Materialize: + +* **Materialize remembers your position.** You report progress with + [`ACKNOWLEDGE`](/sql/acknowledge/), and Materialize stores that position + durably. Your application does not need to persist a timestamp, which + matters for clients that have nowhere durable to put one, such as a browser + or an edge function. + +* **Materialize retains exactly the history you need.** The acknowledged + position, not a fixed duration, determines how much history is kept. You do + not have to guess a history retention period that is long enough to cover + an outage but short enough to afford. + +The trade for that convenience is delivery semantics. A durable subscription is +**at-least-once**: after you reconnect, you may see updates you already +processed. See [Delivery semantics](#delivery-semantics). + +## Syntax + +```mzsql +CREATE DURABLE SUBSCRIPTION ON +WITH (ACKNOWLEDGE WITHIN ) +; + +CREATE DURABLE SUBSCRIPTION +WITH (ACKNOWLEDGE WITHIN ) AS +SELECT FROM [WHERE ] +; +``` + +| Field | Use | +| --- | --- | +| `` | A name for the subscription. Used by [`SUBSCRIBE`](/sql/subscribe/) and [`ACKNOWLEDGE`](/sql/acknowledge/). | +| `` | The source, table, or materialized view to subscribe to. | +| `ACKNOWLEDGE WITHIN` | **Required.** How long you may go without acknowledging. A positive [interval](/sql/types/interval/) value, for example `'1m'`. See [Acknowledgement deadline](#acknowledgement-deadline). | +| `AS SELECT ...` | An optional projection and filter over a single object. See [Supported objects and queries](#supported-objects-and-queries). | + +## Details + +### Starting position + +A new subscription's position is the time at which you created it, and its +acknowledgement deadline starts running from then. Reading it for the first time +therefore gives you the state as of creation, not as of now, so create a +subscription at the point you are ready to start consuming rather than well in +advance. + +### Supported objects and queries + +A durable subscription can target a source, a table, or a materialized view. +Neither views nor indexes are supported, for different reasons: a view stores no +data at all, and an index lives in the memory of a single cluster, so in neither +case is there durable history to retain. + +The `AS SELECT` form accepts a projection and a filter over a **single** object. +Materialize evaluates them while reading the object, so this form builds no +dataflow, adds no compute cost, and resumes exactly as the `ON ` +form does. This covers selecting a subset of columns, computing derived columns, +and filtering rows. + +Anything more is rejected, including joins, aggregations, `DISTINCT`, +subqueries, and more than one object in the `FROM` clause. Temporal filters, +meaning predicates over [`mz_now()`](/sql/functions/now_and_mz_now/), are also +rejected. To subscribe durably to a query of that kind, create a [materialized +view](/sql/create-materialized-view/) for it and target the view. That is also +the faster option at resume time: resuming against a stored collection reads +recent data from storage, whereas subscribing to a query builds a new dataflow +that must rehydrate first. + +A projection can cause distinct rows to become identical, in which case their +changes combine. The result is still a correct stream of changes to the +projected relation. + +If the projection or filter produces an error for some row, for example a +division by zero, the subscription returns that error. Your position is +unchanged, because only [`ACKNOWLEDGE`](/sql/acknowledge/) moves it, so +reconnecting returns the same error until the offending data changes. + +### Acknowledgement deadline + +`ACKNOWLEDGE WITHIN` is the maximum time you may go without acknowledging, +measured on the wall clock: + +* If you acknowledge within it, Materialize guarantees that the + unacknowledged history is still available when you reconnect. + +* If you do not, the subscription **expires**. Materialize stops retaining + history for it, and the next attempt to use it fails with an error rather + than silently skipping the gap. See [Expiry](#expiry). + +Choose a value that covers the outages you expect to recover from, plus the time +your application needs to restart. A minute is a reasonable starting point for +an interactive client. There is a system-wide minimum and maximum; ask your +administrator if you need a value outside them. + +A durable subscription retains history from the moment you create it, whether or +not anything is reading from it. Creating one and never using it retains history +until the deadline expires it. + +Retention has two costs beyond your own storage, both of which argue for keeping +the deadline short: + +* **History is retained for the object according to the furthest behind of + its readers.** If several durable subscriptions target one object, one + reader that stops acknowledging holds history for all of them until its + deadline expires it. + +* **Objects created later must process the retained history.** A + [`CREATE MATERIALIZED VIEW`](/sql/create-materialized-view/), + [`CREATE INDEX`](/sql/create-index/), or [`CREATE SINK`](/sql/create-sink/) + over the same object starts from the oldest time still retained, so a + lagging subscription makes those statements backfill more data. + +### Create one subscription per consumer + +Create a durable subscription per logical consumer, once, and reuse it across +reconnections. It is a provisioned resource, like a table or a view: it is +recorded durably, it appears in the catalog, and creating or dropping one is a +data definition statement. + +Do not create one per page load or per session. Creating and dropping thousands +of short-lived subscriptions is far more expensive than reusing a few long-lived +ones, and every subscription that exists adds to startup work. Give each +consumer a stable name and reconnect to it. + +### Expiry + +When a subscription expires, it is not dropped. It remains visible in +[`mz_internal.mz_durable_subscriptions`](#monitoring), showing the position it +expired at and how long it had gone without acknowledging, so you can tell an +expiry apart from a subscription that never existed. + +Attempting to subscribe using an expired subscription returns a distinguishable +error, so an automated reconnect can branch on it. There are two ways to start +using the subscription again, and both require you to accept that there is a gap +in the data: + +* Subscribe `WITH (RESET)`, which resets the subscription to the current time + and delivers a snapshot there. Use this in a reconnect path: it needs no + privileged statement, and because you asked for it, you know the data you + receive is a fresh snapshot rather than a continuation. It returns an error + if the subscription has not expired, so it cannot displace a live reader by + accident. + +* Run [`ALTER DURABLE SUBSCRIPTION ... RESET`](/sql/alter-durable-subscription/), + which re-arms it at the current time. You then subscribe with `SNAPSHOT true` + to get a usable starting state. This is the operator path. + +{{< note >}} + +Do not try to detect a reset by inspecting the stream. A snapshot arrives as +ordinary insertions at the resume timestamp, so it cannot be told apart from real +inserts at that time. Ask for the reset explicitly, and you know what you got. + +If you do keep your own record of the last position you acknowledged, you can +cross-check it: the first message of any subscribe is a progress message carrying +the timestamp the subscription resumed at. + +{{}} + +### Acknowledging requires progress messages + +Subscribe `WITH (PROGRESS)` whenever you intend to acknowledge. A progress +message is the only thing that tells you a timestamp is complete, and therefore +the only safe thing to acknowledge. Not every timestamp produces a progress +message, and receiving a row at time `t` does not mean that time `t` is +finished, so acknowledging a row's timestamp can skip updates you have not seen. + +### Where reading starts and stops + +By default, omit `AS OF` and Materialize resumes from the position you last +acknowledged, computing the boundary so that you receive updates at and after +that position. This requires nothing from you and is the recommended path. + +You may pass `AS OF` to start somewhere else inside the retained history, which +is useful if you track your own position and are further along than your last +acknowledgement. It behaves exactly as it does for a plain +[`SUBSCRIBE`](/sql/subscribe/#as-of). + +{{< warning >}} + +`AS OF` is an **exclusive** lower bound under `SNAPSHOT false`: `SUBSCRIBE` emits +updates at times *strictly greater* than the timestamp you pass. To receive +updates at time `T`, pass `AS OF T - 1`. Passing `T` silently skips every update +at `T`. + +If you would rather not carry that arithmetic, omit `AS OF` and filter instead. +Every update carries `mz_timestamp`, so a consumer that has committed through `T` +can drop everything below `T` on arrival. Receiving data you already have costs +bandwidth; asking for the wrong starting timestamp costs data. + +{{}} + +`UP TO` bounds where reading stops, exclusively, and is useful for draining a +bounded batch: read `UP TO` a timestamp, commit, acknowledge `UP TO` that same +timestamp, and disconnect. + +`ENVELOPE UPSERT`, `ENVELOPE DEBEZIUM`, and `AS OF AT LEAST` are not supported +on a durable subscription. The envelopes cannot be produced correctly when +resuming without a snapshot, because neither the sink nor Materialize holds the +prior value for a key the resumed stream has not seen. + +### Snapshots + +`SNAPSHOT` defaults to **`false`** here, which is the opposite of a plain +[`SUBSCRIBE`](/sql/subscribe/#snapshot). Resuming without re-snapshotting is the +purpose of a durable subscription, so it is the default. + +A requested snapshot is taken **at your acknowledged position**, not at the +current time, so that the state you receive corresponds to a position you can +name. That gives three ways to reconnect: + +* **Continue**, with `SNAPSHOT false`. You hold state at or past your + acknowledged position and want the changes since. This is the default and + the common case. + +* **Reconcile**, with `SNAPSHOT true`. You know your position but suspect your + local state has drifted. You receive the authoritative state as of your + acknowledged position, which you can compare against what you hold, followed + by subsequent changes. + +* **Start over**, which applies only to a subscription that has + [expired](#expiry). Subscribe `WITH (RESET)`, or run [`ALTER DURABLE + SUBSCRIPTION ... RESET`](/sql/alter-durable-subscription/) and then subscribe + `WITH (SNAPSHOT true)`. Either moves the position to the current time, so the + snapshot is taken there. + +A subscription that has *not* expired cannot be fast-forwarded, because doing so +would discard data it is still holding history for. If you lost your local state +while the subscription is still live, reconcile instead: you receive the state at +your acknowledged position and then replay from there, and that replay is bounded +by `ACKNOWLEDGE WITHIN`. + +### Delivery semantics + +A durable subscription delivers **at least once**. You process a batch, commit +it, and then acknowledge. If your application fails between the commit and the +acknowledgement, Materialize still has the older position, so it re-sends +updates you already processed. + +Your consumer must therefore be idempotent, and the unit of idempotence is the +**timestamp**, not the row. Apply all the updates at one `mz_timestamp` +together, record that you applied it, and skip timestamps you have already +applied. Resuming always starts on a timestamp boundary, so anything re-sent is +re-sent as whole timestamps. + +Do not deduplicate on the combination of `mz_timestamp` and the row. After a +resume the same logical change may arrive consolidated differently than it did +the first time, so per-row comparison cannot distinguish a re-delivery from a +genuine second change. For the same reason, do not treat the stream as an event +log. + +Materialize does not enforce uniqueness, since tables support neither [primary +keys nor unique constraints](/sql/create-table/#known-limitations). If you apply +updates into a keyed store, you are responsible for the key being unique in the +subscribed data. Note that a projection in the `AS SELECT` form can make +distinct rows identical, which combines their changes. + +{{< important >}} + +If you need **exactly-once** processing, keep recording your own progress +timestamp and writing it in the same transaction as the data it covers, as +described in [Resuming +subscriptions](/transform-data/patterns/durable-subscriptions/#note-about-idempotency). + +You can still use a durable subscription for this. Acknowledge as normal to +control how much history is retained, and on reconnection skip what you already +have, either by discarding updates below your own committed timestamp or by +passing `AS OF`. See [Where reading starts and +stops](#where-reading-starts-and-stops) for the trade between the two. + +{{}} + +### Only one reader at a time + +A durable subscription has a single position, so only one reader may use it at a +time. Subscribing again takes over: the new reader starts streaming and the +previous reader's stream fails with an error. + +This is deliberate. It means a client that reconnects does not have to wait for +its own abandoned connection to time out. It also means you should not point two +application instances at the same durable subscription, and if you do, they will +take turns rather than share the work. + +### Changes to the target object + +Dropping the target object fails while a durable subscription exists on it, +unless you use `CASCADE`. + +`SELECT` on the target is checked each time you subscribe, not only when you +create the subscription, so revoking it stops an existing subscription from +being used. + +### Monitoring + +`mz_internal.mz_durable_subscriptions` reports each subscription's acknowledged +position, how long it has gone without acknowledging, how far behind the object +it is, and whether it has expired. Use it to find subscriptions that are holding +history: + +```mzsql +SELECT name, target, acknowledged_up_to, time_since_ack, lag, state +FROM mz_internal.mz_durable_subscriptions +ORDER BY time_since_ack DESC; +``` + +`time_since_ack` is what `ACKNOWLEDGE WITHIN` is compared against, so it is the +column that predicts an expiry. `lag` is the distance to the object's current +time, which is what predicts how much data a reconnection will replay. + +This relation is distinct from +[`mz_internal.mz_subscriptions`](/reference/system-catalog/mz_internal/#mz_subscriptions), +which lists `SUBSCRIBE` statements that are running right now. A durable +subscription appears in `mz_durable_subscriptions` whether or not anything is +currently reading from it. + +## Examples + +### Create a subscription + +```mzsql +CREATE DURABLE SUBSCRIPTION winning_bids_feed +ON winning_bids +WITH (ACKNOWLEDGE WITHIN '1m'); +``` + +### Read from it the first time + +Request a snapshot to bootstrap your application, and `PROGRESS` so you can tell +when a timestamp is complete: + +```mzsql +BEGIN; +DECLARE c CURSOR FOR + SUBSCRIBE USING DURABLE SUBSCRIPTION winning_bids_feed + WITH (PROGRESS, SNAPSHOT true); +``` + +Then loop, buffering each batch until a progress message arrives: + +```mzsql +FETCH ALL c WITH (timeout = '1s'); +``` + +```nofmt + mz_timestamp | mz_progressed | mz_diff | auction_id | amount +---------------+---------------+---------+------------+-------- + 1723459199000 | f | 1 | 1 | 42 + 1723459199000 | f | 1 | 2 | 67 + 1723459200000 | t | | | +``` + +The final row has `mz_progressed` set to `true`, so everything before +`1723459200000` is complete. Process the two buffered updates, commit them, and +then acknowledge that timestamp on the same connection: + +```mzsql +ACKNOWLEDGE DURABLE SUBSCRIPTION winning_bids_feed UP TO 1723459200000; +``` + +Continue fetching and acknowledging for as long as you want to consume. When you +are done, close the cursor and end the transaction: + +```mzsql +CLOSE c; +COMMIT; +``` + +You do not need to acknowledge every progress message. Acknowledging less often +reduces round trips, at the cost of re-processing more data after a failure and +retaining more history in the meantime. + +### Resume after a disconnection + +Reconnect and subscribe again, this time without a snapshot. You supply no +timestamp, because Materialize has your position: + +```mzsql +BEGIN; +DECLARE c CURSOR FOR + SUBSCRIBE USING DURABLE SUBSCRIPTION winning_bids_feed + WITH (PROGRESS); +``` + +`SNAPSHOT` defaults to `false` on this form, so the stream continues rather than +re-snapshotting. The first message you receive is a progress message carrying +the timestamp the subscription resumed at, so your application can confirm it +matches what it expected. + +If your application also lost its local state, request `SNAPSHOT true` to +receive the state as of your acknowledged position, or +[`RESET`](/sql/alter-durable-subscription/) the subscription to start from the +current time instead. + +### Acknowledge from another connection + +If you read with a streaming `SUBSCRIBE` rather than `DECLARE` and `FETCH`, the +PostgreSQL protocol does not allow you to send anything on that connection while +results are streaming. Acknowledge from a second connection instead. A +subscription is addressed by name, so no coordination between the connections is +needed: + +```mzsql +ACKNOWLEDGE DURABLE SUBSCRIPTION winning_bids_feed UP TO 1723459200000; +``` + +## Privileges + +The privileges required to execute this statement are: + +{{% include-headless "/headless/sql-command-privileges/create-durable-subscription" %}} + +## Related pages + +* [`ACKNOWLEDGE`](/sql/acknowledge/) +* [`ALTER DURABLE SUBSCRIPTION`](/sql/alter-durable-subscription/) +* [`DROP DURABLE SUBSCRIPTION`](/sql/drop-durable-subscription/) +* [`SUBSCRIBE`](/sql/subscribe/) +* [`CREATE MATERIALIZED VIEW`](/sql/create-materialized-view/) +* [Resuming subscriptions](/transform-data/patterns/durable-subscriptions/) diff --git a/doc/user/content/sql/drop-durable-subscription.md b/doc/user/content/sql/drop-durable-subscription.md new file mode 100644 index 0000000000000..181af8589fca2 --- /dev/null +++ b/doc/user/content/sql/drop-durable-subscription.md @@ -0,0 +1,60 @@ +--- +title: "DROP DURABLE SUBSCRIPTION" +description: "`DROP DURABLE SUBSCRIPTION` removes a durable subscription and releases the history it retains." +menu: + main: + parent: 'commands' +--- + +{{< private-preview />}} + +`DROP DURABLE SUBSCRIPTION` removes a [durable +subscription](/sql/create-durable-subscription/) and releases the history it was +retaining. + +## Syntax + +```mzsql +DROP DURABLE SUBSCRIPTION [IF EXISTS] +; +``` + +| Field | Use | +| --- | --- | +| `IF EXISTS` | Do not return an error if the subscription does not exist. | +| `` | The durable subscription to remove. | + +## Details + +Dropping a durable subscription releases its hold on the target's history +immediately. Any reader currently streaming from it fails with an error. + +The position is gone. Recreating a subscription with the same name gives you a +new one positioned at the current time, not the one you dropped, so the next read +needs `SNAPSHOT true`. To keep a subscription's identity and privileges while +moving it to the current time, use [`ALTER DURABLE SUBSCRIPTION ... +RESET`](/sql/alter-durable-subscription/) instead. + +Dropping the subscription is also how you stop paying for retained history that +nobody is reading. A subscription that is merely idle continues to retain +history until its acknowledgement deadline expires it. + +## Examples + +```mzsql +DROP DURABLE SUBSCRIPTION winning_bids_feed; +``` + +## Privileges + +The privileges required to execute this statement are: + +{{% include-headless "/headless/sql-command-privileges/drop-durable-subscription" %}} + +## Related pages + +* [`CREATE DURABLE SUBSCRIPTION`](/sql/create-durable-subscription/) +* [`ALTER DURABLE SUBSCRIPTION`](/sql/alter-durable-subscription/) +* [`ACKNOWLEDGE`](/sql/acknowledge/) +* [`SUBSCRIBE`](/sql/subscribe/) +* [Resuming subscriptions](/transform-data/patterns/durable-subscriptions/) diff --git a/doc/user/content/sql/subscribe.md b/doc/user/content/sql/subscribe.md index 52dfea735811f..786cf81726fbd 100644 --- a/doc/user/content/sql/subscribe.md +++ b/doc/user/content/sql/subscribe.md @@ -38,6 +38,13 @@ SUBSCRIBE [TO] [UP TO ] ; +SUBSCRIBE USING DURABLE SUBSCRIPTION +[WITHIN TIMESTAMP ORDER BY [ASC | DESC] [NULLS LAST | NULLS FIRST], ...] +[WITH ( [= ], ...)] +[AS OF ] +[UP TO ] +; + ``` where: @@ -66,7 +73,7 @@ The following options are valid within the `WITH` clause. | Option name | Value type | Default | Describes | | ----------- | ---------- | ------- | --------------------------------------------------------------------------------------------------------------------------------- | -| `SNAPSHOT` | `boolean` | `true` | Whether to emit a snapshot of the current state of the relation at the start of the operation. See [`SNAPSHOT`](#snapshot). | +| `SNAPSHOT` | `boolean` | `true`, but `false` for `USING DURABLE SUBSCRIPTION` | Whether to emit a snapshot of the current state of the relation at the start of the operation. See [`SNAPSHOT`](#snapshot). | | `PROGRESS` | `boolean` | `false` | Whether to include detailed progress information. See [`PROGRESS`](#progress). | ## Details @@ -164,7 +171,12 @@ The value in the `UP TO` clause is automatically [cast to `mz_timestamp`](../../ ### Interaction of `AS OF` and `UP TO` -The lower timestamp bound specified by `AS OF` is inclusive, whereas the upper bound specified by `UP TO` is exclusive. Thus, a `SUBSCRIBE` query whose `AS OF` is equal to its `UP TO` will terminate after returning zero rows. +The lower timestamp bound specified by `AS OF` is inclusive when a snapshot is +emitted, and **exclusive** under [`WITH (SNAPSHOT = false)`](#snapshot), where +`SUBSCRIBE` emits only updates at times strictly greater than the `AS OF` +timestamp. The upper bound specified by `UP TO` is always exclusive. Thus, a +`SUBSCRIBE` query whose `AS OF` is equal to its `UP TO` will terminate after +returning zero rows. A `SUBSCRIBE` whose `UP TO` is less than its `AS OF` timestamp (whether that timestamp was specified in an `AS OF` clause or chosen by the system) will @@ -195,6 +207,11 @@ consists of a series of updates at its [`AS OF`](#as-of) timestamp describing th contents of the relation. After the snapshot, `SUBSCRIBE` emits further updates as they occur. +This default is inverted for `SUBSCRIBE USING DURABLE SUBSCRIPTION`, where +`SNAPSHOT` defaults to `false`, and a requested snapshot is taken at the +subscription's acknowledged position rather than at the current time. See +[`CREATE DURABLE SUBSCRIPTION`](/sql/create-durable-subscription/#snapshots). + For updates in the snapshot, the `mz_timestamp` field will be fast-forwarded to the `AS OF` timestamp. For example, an insert that occurred before the `SUBSCRIBE` began would appear in the snapshot. @@ -578,14 +595,30 @@ DROP SOURCE auction CASCADE; ### Durable subscriptions Because `SUBSCRIBE` requests happen over the network, these connections might -get disrupted for both expected and unexpected reasons. You can adjust the -[history retention +get disrupted for both expected and unexpected reasons. There are two ways to +recover without re-snapshotting. + +The `SUBSCRIBE USING DURABLE SUBSCRIPTION ` form reads from a [`CREATE +DURABLE SUBSCRIPTION`](/sql/create-durable-subscription/) object, which means +Materialize tracks your position for you. You report progress with +[`ACKNOWLEDGE`](/sql/acknowledge/), Materialize retains exactly the history you +have not acknowledged, and you resume without supplying a timestamp. `SNAPSHOT` +defaults to `false` on this form, and a requested snapshot is taken at your +acknowledged position. `ENVELOPE UPSERT`, `ENVELOPE DEBEZIUM`, and `AS OF AT +LEAST` are not supported on this form. Only one reader may use a durable +subscription at a time, and subscribing again takes over from the previous +reader. + +Alternatively, you can adjust the [history retention period](/transform-data/patterns/durable-subscriptions/#history-retention-period) -for the objects a subscription depends on, and then use [`AS OF`](#as-of) to -pick up where you left off on connection drops—this ensures that no data is lost -in the subscription process, and avoids the need for re-snapshotting the data. - -For more information, see [durable +for the objects a subscription depends on, record the progress timestamp in your +own application, and then use [`AS OF`](#as-of) to pick up where you left off. +This requires more from your application, but recording the position yourself is +what makes exactly-once processing possible, because you can write it in the same +transaction as the data. A durable subscription supplies retention; your own +recorded position supplies exactly-once. + +For more information on both, see [resuming subscriptions](/transform-data/patterns/durable-subscriptions/). ## Privileges diff --git a/doc/user/content/transform-data/patterns/durable-subscriptions.md b/doc/user/content/transform-data/patterns/durable-subscriptions.md index 6549ae70028c2..3e15f23d17f41 100644 --- a/doc/user/content/transform-data/patterns/durable-subscriptions.md +++ b/doc/user/content/transform-data/patterns/durable-subscriptions.md @@ -1,6 +1,6 @@ --- -title: "Durable subscriptions" -description: "How to enable lossless, durable subscriptions to your changing results in Materialize" +title: "Resuming subscriptions" +description: "How to resume a subscription after a connection drop without losing or re-snapshotting data" menu: main: parent: 'sql-patterns' @@ -18,15 +18,42 @@ the network, subscriptions might get disrupted for both expected and unexpected reasons. In such cases, it can be useful to have a mechanism to gracefully recover data processing. -To avoid the need for re-processing data that was already sent to your external -application following a connection disruption, you can: - -- Adjust the [history retention period](#history-retention-period) for the - objects that a subscription depends on, and - -- [Access past versions of this - data](#enabling-durable-subscriptions-in-your-application) at specific points - in time to pick up data processing where you left off. +To avoid re-processing data that was already sent to your external application +following a connection disruption, you have two options: let Materialize track +your position with a [durable +subscription](/sql/create-durable-subscription/), or track it yourself by +adjusting the [history retention period](#history-retention-period) and +[accessing past versions of the +data](#enabling-durable-subscriptions-in-your-application) at the point you left +off. + +## Choosing an approach + +There are two ways to recover a subscription, and they differ in who keeps track +of your position. + +- A [`CREATE DURABLE SUBSCRIPTION`](/sql/create-durable-subscription/) object + makes Materialize keep track. You acknowledge what you have processed, + Materialize retains exactly the history you still need, and you resume without + supplying a timestamp. Use this when your application has nowhere durable to + record a timestamp, such as a browser or an edge function, or when you do not + want to guess a history retention period. Delivery is at-least-once, so your + consumer must be idempotent. + +- The pattern described on the rest of this page makes **your application** keep + track. You set a history retention period, record the progress timestamp + yourself, and pass it back as `AS OF` when you resume. This is more work, and + it requires choosing a retention period up front, but it is the only way to + achieve **exactly-once** processing, because it lets you commit the data and + the position in a single transaction. See [Note about + idempotency](#note-about-idempotency). + +The two compose. You can create a durable subscription for its retention +guarantee while still recording your own position: acknowledge to control how +much history is kept, and then either pass your own `AS OF` when you resume, or +omit it and discard every update below your committed timestamp. Filtering +cannot lose data and costs bandwidth; `AS OF` costs nothing and loses data +silently if you forget that it is an exclusive bound. ## History retention period diff --git a/doc/user/data/sql_commands_all.yml b/doc/user/data/sql_commands_all.yml index 700b1bdb6c3cd..5d09c0175ff08 100644 --- a/doc/user/data/sql_commands_all.yml +++ b/doc/user/data/sql_commands_all.yml @@ -3,6 +3,11 @@ columns: - column: object - column: labels rows: + - command: "[`ACKNOWLEDGE`](/sql/acknowledge)" + object: "Durable Subscription" + labels: + - "other" + - "output" - command: "[`ALTER CLUSTER`](/sql/alter-cluster)" object: "Cluster" labels: @@ -29,6 +34,11 @@ rows: labels: - "object" - "owner" + - command: "[`ALTER DURABLE SUBSCRIPTION`](/sql/alter-durable-subscription)" + object: "Durable Subscription" + labels: + - "object" + - "owner" - command: "[`ALTER INDEX`](/sql/alter-index)" object: "Index" labels: @@ -126,6 +136,10 @@ rows: object: "Database" labels: - "object" + - command: "[`CREATE DURABLE SUBSCRIPTION`](/sql/create-durable-subscription)" + object: "Durable Subscription" + labels: + - "object" - command: "[`CREATE INDEX`](/sql/create-index)" object: "Index" labels: @@ -204,6 +218,10 @@ rows: object: "Database" labels: - "object" + - command: "[`DROP DURABLE SUBSCRIPTION`](/sql/drop-durable-subscription)" + object: "Durable Subscription" + labels: + - "object" - command: "[`DROP INDEX`](/sql/drop-index)" object: "Index" labels: