From bd3efaed630214b8e394a0de0de3b3710e390195 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 15 Oct 2026 19:57:04 -0700 Subject: [PATCH 1/3] Held unarchived PostgreSQL records for the archive. The archive tier now composes with the PostgreSQL store through TieredStore. A topic's first record makes the archive's hold, so the janitor drops only what the archive holds. --- crates/sdk-core/CHANGELOG.md | 3 + crates/streams/docs/archive.md | 9 +- crates/streams/docs/postgres.md | 13 ++ crates/streams/src/postgres/log.rs | 23 ++- crates/streams/src/postgres/migrations.rs | 6 + .../postgres/migrations/0003_archive_hold.sql | 171 ++++++++++++++++++ crates/streams/src/postgres/mod.rs | 92 +++++++++- crates/streams/src/service.rs | 3 +- crates/streams/tests/archive_sink.rs | 49 ++++- crates/streams/tests/conformance.rs | 67 +++++++ 10 files changed, 422 insertions(+), 14 deletions(-) create mode 100644 crates/streams/src/postgres/migrations/0003_archive_hold.sql diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 6ca3f115a6..33f95676a0 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -89,6 +89,9 @@ relevant information. first entry. `listen_url` names a direct connection for the listener and the migrations when `url` goes through PgBouncer in transaction mode. `notify_reads` sends a `NOTIFY` per append, so waiting reads in other processes wake at once. +* Experimental: `StreamStoreConfig.archive` works with the PostgreSQL store. A topic's first record + makes the archive's hold as a `floors` row, and the janitor drops only what the archive holds + while the driver keeps up. The archive is off unless the config sets it. * Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the notifications still out. A process calls it before it stops. * Experimental: a server without the stream notifier answers the first notifier call with diff --git a/crates/streams/docs/archive.md b/crates/streams/docs/archive.md index e3e3d5fda1..4c3ff37d16 100644 --- a/crates/streams/docs/archive.md +++ b/crates/streams/docs/archive.md @@ -112,7 +112,8 @@ promise about the order. ## Tests -`tests/archive_sink.rs` runs the driver's sink on memory, on Redis with `STREAMS_REDIS_URL`, and on -S3 with `STREAMS_S3_URL`. The conformance suite runs the archive cases on a local directory, and -on S3 with `STREAMS_S3_URL`, such as `s3://archive/tests` on a local S3-compatible server with -`AWS_ENDPOINT_URL` and `AWS_ALLOW_HTTP=true`. +`tests/archive_sink.rs` runs the driver's sink on memory, on Redis with `STREAMS_REDIS_URL`, on +PostgreSQL with `STREAMS_POSTGRES_URL`, and on S3 with `STREAMS_S3_URL`. The conformance suite runs +the archive cases on a local directory, and on S3 with `STREAMS_S3_URL`, such as +`s3://archive/tests` on a local S3-compatible server with `AWS_ENDPOINT_URL` and +`AWS_ALLOW_HTTP=true`. diff --git a/crates/streams/docs/postgres.md b/crates/streams/docs/postgres.md index 3b4101a5d3..1970a972c7 100644 --- a/crates/streams/docs/postgres.md +++ b/crates/streams/docs/postgres.md @@ -90,6 +90,19 @@ sequence, and the dedupe answers it. Another error reply means the server rolled so it reports `REFUSED`. A read reports `STORAGE`. A server below PostgreSQL 14 is refused as `UNSUPPORTED` at connect. +**Archive hold.** With `archive_lease` set, a driver copies each topic into an archive, and the +store holds what the archive doesn't hold yet. A topic's first record makes the `floors` row +`archive` at position `0`, in the append's transaction, so it holds the whole topic until the +archive's first pass. `advance_archive` moves it to the archive's newest position with no minimum, +renews its lease and sets the topic's `archived`. The janitor reads it as any other hold. When the +driver stops for longer than the lease, the hold ends and retention goes on. `archive_pending` +marks the topics with records to archive, set by appends and closes, and cleared by an advance to +the topic's newest position and by the archive's release. `archive_marks` reads it, closed topics +first. A topic made again after its tombstone starts above every archived position, so an archive +segment name never stands for two records. The archive is optional on PostgreSQL, since records +already live on disk. It helps above about 1 TB of records, for retention in years, and to keep +backups small. An archive needs `notification_log`. + ## Migrations The store embeds its migrations and applies the missing ones at connect, each in its own diff --git a/crates/streams/src/postgres/log.rs b/crates/streams/src/postgres/log.rs index 29ff7f5849..99ef16ae20 100644 --- a/crates/streams/src/postgres/log.rs +++ b/crates/streams/src/postgres/log.rs @@ -2,7 +2,7 @@ use super::{Inner, PARTITIONS, errors::Call, listen::Rings}; use crate::{ - NotificationEntry, NotificationLog, StreamResult, + ArchiveMark, NotificationEntry, NotificationLog, StreamResult, proto::{ChainId, StreamOwnerKind}, }; use std::{sync::Arc, time::Duration}; @@ -11,6 +11,8 @@ pub(super) struct PostgresLog { pub(super) inner: Arc, pub(super) rings: Arc, pub(super) doorbells: u32, + /// Whether the store keeps an archive hold, so it has topics to archive. + pub(super) archive: bool, } /// The chain of a `chains` row. A Workflow chain keeps no owner kind, as the Redis store's keys @@ -113,6 +115,25 @@ impl NotificationLog for PostgresLog { .map(drop) } + async fn archive_marks(&self, max: usize) -> StreamResult> { + if !self.archive || max == 0 { + return Ok(vec![]); + } + let inner = &self.inner; + let max = i64::try_from(max).unwrap_or(i64::MAX); + let rows = inner + .query(Call::Read, &inner.sql.archive_marks, &[&max]) + .await?; + Ok(rows + .into_iter() + .map(|row| ArchiveMark { + chain: chain(row.get(0), row.get(1), row.get(2), row.get(3), row.get(4)), + topic: row.get(5), + written_ms: row.get::<_, i64>(6).max(0) as u64, + }) + .collect()) + } + async fn busy(&self) -> StreamResult> { let inner = &self.inner; let rows = inner.query(Call::Read, &inner.sql.busy, &[]).await?; diff --git a/crates/streams/src/postgres/migrations.rs b/crates/streams/src/postgres/migrations.rs index d59442906f..5e0d64aab2 100644 --- a/crates/streams/src/postgres/migrations.rs +++ b/crates/streams/src/postgres/migrations.rs @@ -32,6 +32,12 @@ pub(super) const MIGRATIONS: &[Migration] = &[ sql: include_str!("migrations/0002_notification_log.sql"), compatible_from: 1, }, + Migration { + version: 3, + name: "archive_hold", + sql: include_str!("migrations/0003_archive_hold.sql"), + compatible_from: 1, + }, ]; /// The first key of every advisory lock the store takes. diff --git a/crates/streams/src/postgres/migrations/0003_archive_hold.sql b/crates/streams/src/postgres/migrations/0003_archive_hold.sql new file mode 100644 index 0000000000..3797460bdc --- /dev/null +++ b/crates/streams/src/postgres/migrations/0003_archive_hold.sql @@ -0,0 +1,171 @@ +-- The archive's hold: a `floors` row with the holder `archive`, made at a topic's first record and +-- moved by the archive, and `archive_pending`, the archive's work list. + +-- As in migration 2, with the archive's hold made at a topic's first record, so retention can't +-- drop a record before the archive's first pass. +CREATE OR REPLACE FUNCTION append_v1( + p_namespace text, p_kind smallint, p_owner text, p_run text, p_activity text, + p_topic text, p_partition smallint, p_producer text, p_attempt bigint, p_sequence bigint, + p_digest bytea, p_records bytea[], p_keeps_topics boolean, p_log boolean, p_doorbells int, + p_archive_lease_ms bigint, p_retention_ms bigint, p_notify_reads boolean, + OUT status smallint, OUT first_position bigint, OUT last_position bigint, + OUT next_sequence bigint, OUT close bytea) +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +DECLARE + v_chain bigint; + v_chain_closed boolean; + v_topic topics%ROWTYPE; + v_held producers%ROWTYPE; + v_count int := cardinality(p_records); + v_now timestamptz; + v_last bigint; +BEGIN + SELECT l.chain_id, l.closed INTO v_chain, v_chain_closed + FROM live_chain_v1(p_namespace, p_kind, p_owner, p_run, p_activity) l; + LOOP + SELECT * INTO v_topic FROM topics t + WHERE t.chain_id = v_chain AND t.topic = p_topic FOR NO KEY UPDATE; + EXIT WHEN FOUND; + -- Checked before the topic exists, so a refused batch leaves no trace of it. + IF v_chain_closed THEN + status := 4; + RETURN; + END IF; + PERFORM topic_id_v1(v_chain, p_topic, p_partition); + END LOOP; + SELECT * INTO v_held FROM producers h + WHERE h.topic_id = v_topic.topic_id AND h.producer_id = p_producer + AND h.attempt = p_attempt; + IF FOUND THEN + IF p_sequence = v_held.sequence THEN + IF p_digest = v_held.digest THEN + status := 1; + first_position := v_held.first_position; + last_position := v_held.first_position + v_held.count - 1; + ELSE + status := 2; + END IF; + RETURN; + END IF; + IF p_sequence < v_held.sequence + v_held.count THEN + status := 3; + next_sequence := v_held.sequence + v_held.count; + RETURN; + END IF; + END IF; + IF v_chain_closed OR v_topic.closed THEN + status := 4; + close := v_topic.close; + RETURN; + END IF; + IF p_archive_lease_ms IS NOT NULL AND v_topic.last_position IS NULL THEN + INSERT INTO floors + VALUES (v_topic.topic_id, 'archive', 0, now() + p_archive_lease_ms * interval '1 millisecond') + ON CONFLICT DO NOTHING; + END IF; + -- Taken under the row lock, so the times grow with the positions. + v_now := clock_timestamp(); + v_last := coalesce(v_topic.last_position, v_topic.base); + INSERT INTO records (topic_id, position, written_at, record) + SELECT v_topic.topic_id, v_last + u.i, v_now, u.r + FROM unnest(p_records) WITH ORDINALITY AS u(r, i); + INSERT INTO producers VALUES (v_topic.topic_id, p_producer, p_attempt, p_sequence, v_count, + v_last + 1, p_digest, v_now) + ON CONFLICT (topic_id, producer_id, attempt) DO UPDATE + SET sequence = excluded.sequence, count = excluded.count, + first_position = excluded.first_position, digest = excluded.digest, + written_at = excluded.written_at; + UPDATE topics t SET + last_position = v_last + v_count, + written_at = v_now, + written = t.written OR (p_keeps_topics AND p_sequence = 1), + archive_pending = t.archive_pending OR p_archive_lease_ms IS NOT NULL, + trim_at = coalesce(t.trim_at, v_now + p_retention_ms * interval '1 millisecond') + WHERE t.topic_id = v_topic.topic_id; + IF p_log THEN + -- A busy partition rings no more, so rings stay rare under load. + IF NOT EXISTS (SELECT 1 FROM notifications n WHERE n.partition = v_topic.partition) THEN + PERFORM pg_notify(current_schema() || '_bell_' || (v_topic.partition % p_doorbells), + v_topic.partition::text); + END IF; + INSERT INTO notifications (partition, topic_id, position) + VALUES (v_topic.partition, v_topic.topic_id, v_last + v_count); + END IF; + IF p_notify_reads THEN + PERFORM pg_notify(current_schema() || '_read_' || (v_topic.topic_id % 64), + v_topic.topic_id::text); + END IF; + status := 0; + first_position := v_last + 1; + last_position := v_last + v_count; +END $$; + +-- As in migration 1. A close with `p_archive` puts the topic on the archive's work list, first, +-- since a closed topic gets nothing new. +CREATE OR REPLACE FUNCTION close_topic_v1(p_namespace text, p_kind smallint, p_owner text, + p_run text, p_activity text, p_topic text, p_partition smallint, p_close bytea, + p_archive boolean) +RETURNS bytea +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +DECLARE + v_chain bigint; + v_topic bigint; + v_close bytea; +BEGIN + SELECT l.chain_id INTO v_chain + FROM live_chain_v1(p_namespace, p_kind, p_owner, p_run, p_activity) l; + LOOP + v_topic := topic_id_v1(v_chain, p_topic, p_partition); + UPDATE topics t SET closed = true, close = coalesce(t.close, p_close), + archive_pending = t.archive_pending OR p_archive + WHERE t.topic_id = v_topic + RETURNING t.close INTO v_close; + EXIT WHEN FOUND; + END LOOP; + RETURN v_close; +END $$; + +-- As in migration 1. The archive releases once it holds a closed topic to its end, so the topic +-- also leaves the archive's work list. +CREATE OR REPLACE FUNCTION release_floor_v1(p_namespace text, p_kind smallint, p_owner text, + p_run text, p_activity text, p_topic text, p_holder text, p_grace_ms bigint) RETURNS void +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +BEGIN + UPDATE floors f SET until = least(f.until, now() + p_grace_ms * interval '1 millisecond') + FROM chains c JOIN topics t ON t.chain_id = c.chain_id AND t.topic = p_topic + WHERE (c.namespace, c.owner_kind, c.owner_id, c.first_run_id, c.activity_id) + = (p_namespace, p_kind, p_owner, p_run, p_activity) + AND f.topic_id = t.topic_id AND f.holder = p_holder; + IF p_holder = 'archive' THEN + UPDATE topics t SET archive_pending = false FROM chains c + WHERE (c.namespace, c.owner_kind, c.owner_id, c.first_run_id, c.activity_id) + = (p_namespace, p_kind, p_owner, p_run, p_activity) + AND t.chain_id = c.chain_id AND t.topic = p_topic; + END IF; +END $$; + +-- Moves the archive's hold to `p_position` with no minimum, renews its lease and keeps the +-- position as the topic's `archived`. A topic with no row stays without one. +CREATE FUNCTION advance_archive_v1(p_namespace text, p_kind smallint, p_owner text, p_run text, + p_activity text, p_topic text, p_position bigint, p_lease_ms bigint) RETURNS void +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +DECLARE + v_topic bigint; +BEGIN + -- The key share lock keeps the janitor from deleting the topic under the insert. + SELECT t.topic_id INTO v_topic FROM chains c + JOIN topics t ON t.chain_id = c.chain_id AND t.topic = p_topic + WHERE (c.namespace, c.owner_kind, c.owner_id, c.first_run_id, c.activity_id) + = (p_namespace, p_kind, p_owner, p_run, p_activity) + FOR KEY SHARE OF t; + IF NOT FOUND THEN + RETURN; + END IF; + INSERT INTO floors AS f + VALUES (v_topic, 'archive', p_position, now() + p_lease_ms * interval '1 millisecond') + ON CONFLICT (topic_id, holder) DO UPDATE + SET position = excluded.position, until = excluded.until; + UPDATE topics t SET archived = p_position, + archive_pending = t.archive_pending AND t.last_position IS DISTINCT FROM p_position + WHERE t.topic_id = v_topic; +END $$; diff --git a/crates/streams/src/postgres/mod.rs b/crates/streams/src/postgres/mod.rs index efb1891fee..7178c9eca6 100644 --- a/crates/streams/src/postgres/mod.rs +++ b/crates/streams/src/postgres/mod.rs @@ -82,6 +82,10 @@ pub struct PostgresStoreOptions { /// Each append also sends `NOTIFY`, so waiting reads in other processes wake in milliseconds. /// Costs append throughput when many producers write at once. pub notify_reads: bool, + /// With an archive, how long the archive's hold on a topic lasts after the archive last moved + /// it. Each topic's first record makes the hold, so retention drops only what the archive + /// holds while the archive keeps up. Needs `notification_log`, since a driver archives. + pub archive_lease: Option, } impl PostgresStoreOptions { @@ -102,6 +106,7 @@ impl PostgresStoreOptions { notify_doorbells: DEFAULT_POSTGRES_DOORBELLS, listen_url: None, notify_reads: false, + archive_lease: None, } } } @@ -205,6 +210,8 @@ struct Sql { ack: String, busy: String, cap_log: String, + advance: String, + archive_marks: String, } /// The chain key, as `$1` to `$5`. @@ -291,6 +298,19 @@ impl Sql { WHERE EXISTS (SELECT 1 FROM {s}.notifications n WHERE n.partition = p)" ), cap_log: format!("SELECT {s}.cap_log_v1($1)"), + advance: format!("SELECT {s}.advance_archive_v1($1, $2, $3, $4, $5, $6, $7, $8)"), + archive_marks: format!( + "SELECT x.namespace, x.owner_kind, x.owner_id, x.first_run_id, x.activity_id, \ + x.topic, x.written_ms FROM ( \ + SELECT c.namespace, c.owner_kind, c.owner_id, c.first_run_id, c.activity_id, \ + t.topic, CASE WHEN t.closed THEN 0 \ + ELSE (extract(epoch FROM t.written_at) * 1000)::bigint END AS written_ms, \ + row_number() OVER (PARTITION BY t.partition \ + ORDER BY t.closed DESC, t.written_at) AS n \ + FROM {s}.topics t JOIN {s}.chains c ON c.chain_id = t.chain_id \ + WHERE t.archive_pending) x \ + WHERE x.n <= $1" + ), } } } @@ -311,6 +331,8 @@ struct Inner { log: bool, doorbells: i32, notify_reads: bool, + /// The archive's lease in milliseconds, with an archive. + archive_lease_ms: Option, } impl PostgresStore { @@ -327,6 +349,11 @@ impl PostgresStore { options.schema ))); } + if options.archive_lease.is_some() && !options.notification_log { + return Err(StreamError::refused( + "an archive needs the notification log, since a driver archives", + )); + } if !(1..=PARTITIONS).contains(&options.notify_doorbells) { return Err(StreamError::refused(format!( "the notification log rings 1 to {PARTITIONS} doorbells, not {}", @@ -381,6 +408,7 @@ impl PostgresStore { log: options.notification_log, doorbells: options.notify_doorbells as i32, notify_reads: options.notify_reads, + archive_lease_ms: options.archive_lease.map(|lease| millis(lease).max(1)), }); let mut tasks = vec![ tokio::spawn(poll(Arc::downgrade(&inner), options.poll_interval)), @@ -401,6 +429,7 @@ impl PostgresStore { inner: inner.clone(), rings, doorbells: options.notify_doorbells, + archive: options.archive_lease.is_some(), }) }); Ok(Self { inner, log, tasks }) @@ -693,7 +722,6 @@ impl StreamStore for PostgresStore { } let inner = &self.inner; let chain = chain_of(request.chain.as_ref())?; - let no_lease: Option = None; let rows = inner .query( Call::Append, @@ -714,7 +742,7 @@ impl StreamStore for PostgresStore { &chain.keeps_topics(), &inner.log, &inner.doorbells, - &no_lease, + &inner.archive_lease_ms, &inner.retention_ms, &inner.notify_reads, ], @@ -971,7 +999,7 @@ impl StreamStore for PostgresStore { &topic, &chain.partition(), &close.encode_to_vec(), - &false, + &inner.archive_lease_ms.is_some(), ], ) .await?; @@ -1031,6 +1059,48 @@ impl StreamStore for PostgresStore { Ok(rows.iter().map(|row| row.get(0)).collect()) } + async fn advance_archive( + &self, + chain: &ChainId, + topic: &str, + position: &str, + ) -> StreamResult<()> { + let inner = &self.inner; + let Some(lease) = inner.archive_lease_ms else { + return Err(StreamError::unsupported("this store keeps no archive hold")); + }; + let position = parse_position(position)?.unwrap_or(0); + let chain = Chain::of(chain); + inner + .query( + Call::Write, + &inner.sql.advance, + &[ + &chain.namespace, + &chain.kind, + &chain.owner, + &chain.run, + &chain.activity, + &topic, + &position, + &lease, + ], + ) + .await + .map(drop) + } + + async fn archived(&self, chain: &ChainId, topic: &str) -> StreamResult> { + if self.inner.archive_lease_ms.is_none() { + return Ok(None); + } + Ok(self + .inner + .topic_row(chain, topic) + .await? + .map(|(_, _, archived)| archived.map(|p| p.to_string()).unwrap_or_default())) + } + async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { let (kind, owner): (i16, &str) = match StreamOwnerKind::try_from(request.owner_kind) { Ok(StreamOwnerKind::Workflow) => (1, &request.workflow_id), @@ -1071,6 +1141,18 @@ impl Inner { chain: &ChainId, topic: &str, ) -> StreamResult<(Option, Option, Option)> { + Ok(self + .topic_row(chain, topic) + .await? + .unwrap_or((None, None, None))) + } + + /// [Inner::topic_state], or `None` when the topic has no row. + async fn topic_row( + &self, + chain: &ChainId, + topic: &str, + ) -> StreamResult, Option, Option)>> { let chain = Chain::of(chain); let rows = self .query( @@ -1086,9 +1168,7 @@ impl Inner { ], ) .await?; - Ok(rows.first().map_or((None, None, None), |row| { - (row.get(0), row.get(1), row.get(2)) - })) + Ok(rows.first().map(|row| (row.get(0), row.get(1), row.get(2)))) } } diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 39e1845844..136d60d5a8 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -439,10 +439,11 @@ async fn connect_redis( async fn connect_postgres( config: crate::proto::PostgresStoreConfig, notification_log: bool, - _archive_lease: Option, + archive_lease: Option, ) -> StreamResult<(Arc, Duration)> { let mut options = crate::PostgresStoreOptions::new(config.url); options.notification_log = notification_log; + options.archive_lease = archive_lease; if config.notify_doorbells > 0 { options.notify_doorbells = config.notify_doorbells; } diff --git a/crates/streams/tests/archive_sink.rs b/crates/streams/tests/archive_sink.rs index fcccf2f4ea..a67d231cf0 100644 --- a/crates/streams/tests/archive_sink.rs +++ b/crates/streams/tests/archive_sink.rs @@ -1,8 +1,9 @@ //! The stream driver's archive sink: it copies the topics a store marks into an archive, so a //! `TieredStore` reads them after the store drops them. //! -//! The memory cases always run. Set `STREAMS_REDIS_URL` for the Redis case, and `STREAMS_S3_URL` -//! with the `AWS_*` variables for the S3 case. +//! The memory cases always run. Set `STREAMS_REDIS_URL` for the Redis case, +//! `STREAMS_POSTGRES_URL` for the PostgreSQL case, and `STREAMS_S3_URL` with the `AWS_*` variables +//! for the S3 case. use std::{ sync::{ @@ -267,3 +268,47 @@ async fn the_driver_archives_a_redis_store_into_a_directory() { .await; std::fs::remove_dir_all(root).unwrap(); } + +#[cfg(feature = "postgres")] +#[tokio::test] +async fn the_driver_archives_a_postgres_store_into_a_directory() { + let Ok(url) = std::env::var("STREAMS_POSTGRES_URL") else { + eprintln!("set STREAMS_POSTGRES_URL to run the PostgreSQL archive cases"); + return; + }; + let schema = unique("archive_sink").replace('-', "_"); + let mut options = temporalio_streams::PostgresStoreOptions::new(url.clone()); + options.schema = schema.clone(); + options.max_connections = 4; + options.notification_log = true; + options.archive_lease = Some(Duration::from_secs(60)); + let store = Arc::new( + temporalio_streams::PostgresStore::connect(options) + .await + .unwrap(), + ); + let root = std::env::temp_dir().join(unique("streams-archive-sink")); + let archive = Arc::new(LocalFileArchive::new(&root)); + let (raw, connection) = tokio_postgres::connect(&url, tokio_postgres::NoTls) + .await + .unwrap(); + tokio::spawn(connection); + let raw = Arc::new(raw); + the_driver_archives_closed_and_busy_topics(store, archive, async |chain, topic| { + // The topic row goes with its records, as when the janitor deleted a tombstone. + raw.execute( + &format!( + "WITH t AS (SELECT t.topic_id FROM {schema}.chains c \ + JOIN {schema}.topics t ON t.chain_id = c.chain_id AND t.topic = $3 \ + WHERE c.namespace = $1 AND c.owner_id = $2), \ + gone AS (DELETE FROM {schema}.records r USING t WHERE r.topic_id = t.topic_id) \ + DELETE FROM {schema}.topics x USING t WHERE x.topic_id = t.topic_id" + ), + &[&chain.namespace, &chain.workflow_id, &topic], + ) + .await + .unwrap(); + }) + .await; + std::fs::remove_dir_all(root).unwrap(); +} diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 75363a5c26..08bbe00059 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -808,6 +808,16 @@ mod on_postgres { .await } + /// [postgres_with_log] with an archive hold, for a tiered store. + pub(crate) async fn postgres_with_archive(variable: &str) -> Option { + postgres_with(variable, |options| { + options.notification_log = true; + options.notify_doorbells = 4; + options.archive_lease = Some(Duration::from_secs(60)); + }) + .await + } + /// [postgres] with a `NOTIFY` per append for waiting reads. pub(crate) async fn postgres_notifying_reads(variable: &str) -> Option { postgres_with(variable, |options| options.notify_reads = true).await @@ -3218,6 +3228,63 @@ mod archive_workflow_activity_cluster_redis { ))); } +#[cfg(feature = "postgres")] +mod archive_postgres { + archive_cases!(super::tiered(super::on_postgres::postgres_with_archive( + "STREAMS_POSTGRES_URL" + ))); + // The test trim keeps what a hold holds, as the janitor does. + archive_conformance!(super::tiered(super::on_postgres::postgres_with_archive( + "STREAMS_POSTGRES_URL" + )); + the_trim_keeps_what_the_archive_does_not_hold, + ); + log_cases!(super::tiered(super::on_postgres::postgres_with_archive( + "STREAMS_POSTGRES_URL" + ))); +} + +#[cfg(feature = "postgres")] +mod tiered_postgres { + // The test trim keeps what the archive's hold holds, so the store cases that trim run with no + // hold, as on memory. + store_cases!(super::tiered(super::on_postgres::postgres_with_log( + "STREAMS_POSTGRES_URL" + ))); + log_cases!(super::tiered(super::on_postgres::postgres_with_log( + "STREAMS_POSTGRES_URL" + ))); + conformance!(super::tiered(super::on_postgres::postgres("STREAMS_POSTGRES_URL")); + deleting_an_owner_drops_every_chain_of_it, + ); +} + +#[cfg(feature = "postgres")] +mod archive_activity_postgres { + archive_cases!(super::tiered(super::activity( + super::on_postgres::postgres_with_archive("STREAMS_POSTGRES_URL") + ))); +} + +#[cfg(feature = "postgres")] +mod archive_workflow_activity_postgres { + archive_cases!(super::tiered(super::workflow_activity( + super::on_postgres::postgres_with_archive("STREAMS_POSTGRES_URL") + ))); +} + +#[cfg(all(feature = "archive", feature = "postgres"))] +mod archive_s3_postgres { + archive_cases!(super::tiered_on_s3( + super::on_postgres::postgres_with_archive("STREAMS_POSTGRES_URL") + )); + archive_conformance!(super::tiered_on_s3(super::on_postgres::postgres_with_archive( + "STREAMS_POSTGRES_URL" + )); + the_trim_keeps_what_the_archive_does_not_hold, + ); +} + #[cfg(feature = "archive")] mod archive_s3_memory { archive_cases!(super::tiered_on_s3(super::memory_holding_for_archive())); From d34b91c9c34449e3e55d65af22d512f5cdb6ab2b Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 05:53:48 -0700 Subject: [PATCH 2/3] Fixed the PostgreSQL janitor, rings, log cap and holds in migration 0004. Held topics took every janitor pass, a ring was lost while a driver held taken entries, the log cap dropped entries a driver worked on, a topic took any number of holders, the archive hold could move back, and a lost first batch kept a topic out of its set. --- crates/streams/docs/postgres.md | 18 +- crates/streams/src/postgres/migrations.rs | 6 + .../postgres/migrations/0004_review_fixes.sql | 260 ++++++++++++++++++ crates/streams/tests/conformance.rs | 16 ++ crates/streams/tests/postgres_store.rs | 168 +++++++++++ 5 files changed, 462 insertions(+), 6 deletions(-) create mode 100644 crates/streams/src/postgres/migrations/0004_review_fixes.sql diff --git a/crates/streams/docs/postgres.md b/crates/streams/docs/postgres.md index 4cd45c9b88..ed6cfaed19 100644 --- a/crates/streams/docs/postgres.md +++ b/crates/streams/docs/postgres.md @@ -52,11 +52,14 @@ at a time across the database. With many producers at once, it costs append thro `notifications` in its transaction: the topic and the append's last position, in one of 64 partitions fixed in the schema. Every topic of a chain lands in the partition that the first four bytes of SHA-256 over the chain's fields give. A repeat of the newest batch writes none. When the -partition held no entry before, the append also sends `NOTIFY` on `_bell_` with the partition, so the doorbell costs nothing while a partition stays busy. Drivers +partition held no entry that a driver hasn't taken, the append also sends `NOTIFY` on +`_bell_` with the partition, so the doorbell costs nothing while a +partition stays busy, and a driver that drains taken entries still hears of the next one. Drivers take entries with `FOR UPDATE SKIP LOCKED`: a consumer's own unacked entries, those another consumer left for `claim_idle`, and new ones. An ack deletes them. The janitor keeps the newest -10,000 entries per partition, so a log with no driver stays bounded. An entry whose topic an owner +10,000 entries per partition, so a log with no driver stays bounded. It keeps older entries that a +driver took within the last minute too, so a quiet stream that shares a partition with a busy one +keeps its only entry while a slow driver works. An entry whose topic an owner delete removed is acked as broken when a driver takes it. **Listener.** One connection per store listens for the doorbells and, with `notify_reads`, the read @@ -69,8 +72,9 @@ lost ring too. The listener sets TCP keepalive to notice a dead connection withi **Retention.** PostgreSQL has no key expiry, so each store runs a janitor every `janitor_interval` (one minute). A transaction advisory lock around each step lets one janitor -work at a time across processes, also through PgBouncer in transaction mode. A pass trims the topics that are due, at most 10,000 records per topic per -transaction, and keeps what a hold holds. It moves the topic's newest dropped position in the same +work at a time across processes, also through PgBouncer in transaction mode. A pass trims the +topics that are due, at most 10,000 records per topic per transaction, and keeps what a hold +holds. It moves the topic's newest dropped position in the same transaction. A topic whose records are all gone keeps its row as the tombstone for the retention plus 30 days after its last write, as Redis keeps a topic's meta. Then the janitor deletes it, and a chain row with no topics that nothing touched for as long. `PostgresStore::run_janitor` runs a @@ -78,7 +82,9 @@ pass at once. **Floors.** `floors` keeps one row per holder of a topic, with the lowest position it held and the end of its lease. Position `0` holds the whole topic. A release moves the end to now plus the -release grace. The janitor deletes ended holds before each trim. +release grace. The janitor deletes ended holds before each trim. A topic keeps at most 1,000 live +holders, and a new holder past them is refused. A topic whose oldest record a hold keeps is due +again when the first hold ends, so held topics don't take every janitor pass. **Owner deletes.** One function deletes an owner's chain rows, their topics and their holds, and puts the topics in `purge`. Then the store deletes their records and producer rows in batches of diff --git a/crates/streams/src/postgres/migrations.rs b/crates/streams/src/postgres/migrations.rs index 5e0d64aab2..429b21425a 100644 --- a/crates/streams/src/postgres/migrations.rs +++ b/crates/streams/src/postgres/migrations.rs @@ -38,6 +38,12 @@ pub(super) const MIGRATIONS: &[Migration] = &[ sql: include_str!("migrations/0003_archive_hold.sql"), compatible_from: 1, }, + Migration { + version: 4, + name: "review_fixes", + sql: include_str!("migrations/0004_review_fixes.sql"), + compatible_from: 1, + }, ]; /// The first key of every advisory lock the store takes. diff --git a/crates/streams/src/postgres/migrations/0004_review_fixes.sql b/crates/streams/src/postgres/migrations/0004_review_fixes.sql new file mode 100644 index 0000000000..d97180a749 --- /dev/null +++ b/crates/streams/src/postgres/migrations/0004_review_fixes.sql @@ -0,0 +1,260 @@ +-- Fixes after review: retention, rings, the log cap, the holder cap and the archive's hold. + +-- As in migration 3. Every batch puts a Workflow Activity's topic in its set, since a first batch +-- can be lost. The doorbell rings unless the partition holds an entry no driver took. +CREATE OR REPLACE FUNCTION append_v1( + p_namespace text, p_kind smallint, p_owner text, p_run text, p_activity text, + p_topic text, p_partition smallint, p_producer text, p_attempt bigint, p_sequence bigint, + p_digest bytea, p_records bytea[], p_keeps_topics boolean, p_log boolean, p_doorbells int, + p_archive_lease_ms bigint, p_retention_ms bigint, p_notify_reads boolean, + OUT status smallint, OUT first_position bigint, OUT last_position bigint, + OUT next_sequence bigint, OUT close bytea) +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +DECLARE + v_chain bigint; + v_chain_closed boolean; + v_topic topics%ROWTYPE; + v_held producers%ROWTYPE; + v_count int := cardinality(p_records); + v_now timestamptz; + v_last bigint; +BEGIN + SELECT l.chain_id, l.closed INTO v_chain, v_chain_closed + FROM live_chain_v1(p_namespace, p_kind, p_owner, p_run, p_activity) l; + LOOP + SELECT * INTO v_topic FROM topics t + WHERE t.chain_id = v_chain AND t.topic = p_topic FOR NO KEY UPDATE; + EXIT WHEN FOUND; + -- Checked before the topic exists, so a refused batch leaves no trace of it. + IF v_chain_closed THEN + status := 4; + RETURN; + END IF; + PERFORM topic_id_v1(v_chain, p_topic, p_partition); + END LOOP; + SELECT * INTO v_held FROM producers h + WHERE h.topic_id = v_topic.topic_id AND h.producer_id = p_producer + AND h.attempt = p_attempt; + IF FOUND THEN + IF p_sequence = v_held.sequence THEN + IF p_digest = v_held.digest THEN + status := 1; + first_position := v_held.first_position; + last_position := v_held.first_position + v_held.count - 1; + ELSE + status := 2; + END IF; + RETURN; + END IF; + IF p_sequence < v_held.sequence + v_held.count THEN + status := 3; + next_sequence := v_held.sequence + v_held.count; + RETURN; + END IF; + END IF; + IF v_chain_closed OR v_topic.closed THEN + status := 4; + close := v_topic.close; + RETURN; + END IF; + IF p_archive_lease_ms IS NOT NULL AND v_topic.last_position IS NULL THEN + INSERT INTO floors + VALUES (v_topic.topic_id, 'archive', 0, now() + p_archive_lease_ms * interval '1 millisecond') + ON CONFLICT DO NOTHING; + END IF; + -- Taken under the row lock, so the times grow with the positions. + v_now := clock_timestamp(); + v_last := coalesce(v_topic.last_position, v_topic.base); + INSERT INTO records (topic_id, position, written_at, record) + SELECT v_topic.topic_id, v_last + u.i, v_now, u.r + FROM unnest(p_records) WITH ORDINALITY AS u(r, i); + INSERT INTO producers VALUES (v_topic.topic_id, p_producer, p_attempt, p_sequence, v_count, + v_last + 1, p_digest, v_now) + ON CONFLICT (topic_id, producer_id, attempt) DO UPDATE + SET sequence = excluded.sequence, count = excluded.count, + first_position = excluded.first_position, digest = excluded.digest, + written_at = excluded.written_at; + UPDATE topics t SET + last_position = v_last + v_count, + written_at = v_now, + written = t.written OR p_keeps_topics, + archive_pending = t.archive_pending OR p_archive_lease_ms IS NOT NULL, + trim_at = coalesce(t.trim_at, v_now + p_retention_ms * interval '1 millisecond') + WHERE t.topic_id = v_topic.topic_id; + IF p_log THEN + -- A partition with entries no driver took yet rings no more, so rings stay rare under load. + -- An entry a driver took doesn't count, since the driver may have finished its drain. + IF NOT EXISTS (SELECT 1 FROM notifications n + WHERE n.partition = v_topic.partition AND n.taken_by IS NULL) THEN + PERFORM pg_notify(current_schema() || '_bell_' || (v_topic.partition % p_doorbells), + v_topic.partition::text); + END IF; + INSERT INTO notifications (partition, topic_id, position) + VALUES (v_topic.partition, v_topic.topic_id, v_last + v_count); + END IF; + IF p_notify_reads THEN + PERFORM pg_notify(current_schema() || '_read_' || (v_topic.topic_id % 64), + v_topic.topic_id::text); + END IF; + status := 0; + first_position := v_last + 1; + last_position := v_last + v_count; +END $$; + +-- As in migration 1. A topic whose hold keeps its oldest record is due again when a hold ends, +-- not at once, so held topics don't take every janitor pass. +CREATE OR REPLACE FUNCTION trim_topic_v1(p_topic bigint, p_retention_ms bigint, p_grace_ms bigint, + p_batch int, OUT deleted bigint, OUT done boolean) +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +DECLARE + t topics%ROWTYPE; + v_retention interval := p_retention_ms * interval '1 millisecond'; + v_grace interval := p_grace_ms * interval '1 millisecond'; + v_expired timestamptz := now() - v_retention; + v_keep bigint; + v_cut bigint; + v_floor bigint; + v_floor_until timestamptz; + v_limit bigint; + v_oldest timestamptz; + v_held boolean; +BEGIN + deleted := 0; + SELECT * INTO t FROM topics WHERE topic_id = p_topic FOR UPDATE SKIP LOCKED; + IF NOT FOUND THEN done := false; RETURN; END IF; + DELETE FROM floors f WHERE f.topic_id = p_topic AND f.until <= now(); + SELECT min(f.position), min(f.until) INTO v_floor, v_floor_until + FROM floors f WHERE f.topic_id = p_topic; + -- `written_at` grows with the position, so this walks only what it deletes, plus one. + SELECT r.position INTO v_keep FROM records r + WHERE r.topic_id = p_topic AND r.written_at >= v_expired ORDER BY r.position LIMIT 1; + v_cut := coalesce(v_keep - 1, t.last_position); + v_held := v_floor IS NOT NULL AND v_floor - 1 < v_cut; + IF v_held THEN v_cut := v_floor - 1; END IF; + SELECT r.position INTO v_limit FROM records r + WHERE r.topic_id = p_topic AND r.position <= v_cut + ORDER BY r.position OFFSET p_batch - 1 LIMIT 1; + done := v_limit IS NULL OR v_limit = v_cut; + v_cut := least(v_cut, v_limit); + IF v_cut > coalesce(t.trimmed, t.base) THEN + DELETE FROM records r WHERE r.topic_id = p_topic AND r.position <= v_cut; + GET DIAGNOSTICS deleted = ROW_COUNT; + IF deleted > 0 THEN t.trimmed := v_cut; END IF; + END IF; + DELETE FROM producers h WHERE h.topic_id = p_topic AND h.written_at < v_expired; + SELECT r.written_at INTO v_oldest FROM records r WHERE r.topic_id = p_topic + ORDER BY r.position LIMIT 1; + IF v_oldest IS NULL AND v_floor IS NULL AND t.archive_pending IS NOT TRUE + AND coalesce(t.written_at, '-infinity') < now() - v_retention - v_grace THEN + -- The tombstone outlived its grace. The time-based base keeps later positions above it. + DELETE FROM producers h WHERE h.topic_id = p_topic; + DELETE FROM topics WHERE topic_id = p_topic; + -- A busy chain keeps its row. A later pass of the idle chains takes it. + IF pg_try_advisory_xact_lock(hashtext(current_schema()), (t.chain_id % 2147483647)::int) THEN + DELETE FROM chains c WHERE c.chain_id = t.chain_id + AND c.touched_at < now() - v_retention - v_grace + AND NOT EXISTS (SELECT 1 FROM topics x WHERE x.chain_id = t.chain_id); + END IF; + done := true; + RETURN; + END IF; + UPDATE topics SET trimmed = t.trimmed, + trim_at = CASE + WHEN NOT done THEN now() + WHEN v_oldest IS NOT NULL AND v_floor IS NULL THEN v_oldest + v_retention + -- The hold keeps the oldest record, so nothing moves before a hold ends. + WHEN v_oldest IS NOT NULL AND v_held THEN v_floor_until + WHEN v_oldest IS NOT NULL THEN least(v_oldest + v_retention, v_floor_until) + WHEN v_floor IS NOT NULL THEN v_floor_until + ELSE coalesce(t.written_at, now()) + v_retention + v_grace END + WHERE topic_id = p_topic; +END $$; + +-- As in migration 1, with at most 1,000 live holders per topic. +CREATE OR REPLACE FUNCTION hold_floor_v1(p_namespace text, p_kind smallint, p_owner text, + p_run text, p_activity text, p_topic text, p_partition smallint, p_holder text, + p_position bigint, p_lease_ms bigint) RETURNS void +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +DECLARE + v_chain bigint; + v_topic bigint; +BEGIN + SELECT l.chain_id INTO v_chain + FROM live_chain_v1(p_namespace, p_kind, p_owner, p_run, p_activity) l; + LOOP + v_topic := topic_id_v1(v_chain, p_topic, p_partition); + -- The key share lock keeps the janitor from deleting the topic under the insert. + PERFORM 1 FROM topics t WHERE t.topic_id = v_topic FOR KEY SHARE; + EXIT WHEN FOUND; + END LOOP; + -- Every trim reads the holds, and each one pins records, so a topic takes at most 1,000 live + -- holders, as `MAX_FLOOR_HOLDERS` says. A holder already on the topic renews. + IF NOT EXISTS (SELECT 1 FROM floors f WHERE f.topic_id = v_topic AND f.holder = p_holder) + AND (SELECT count(*) FROM floors f WHERE f.topic_id = v_topic AND f.until > now()) >= 1000 + THEN + RAISE EXCEPTION 'topic % already has 1000 floor holders, so a positions-only reader can not ' + 'start on it', p_topic USING ERRCODE = 'P0001'; + END IF; + -- Position 0 holds the whole topic. The lowest position a holder held wins. + INSERT INTO floors AS f + VALUES (v_topic, p_holder, p_position, now() + p_lease_ms * interval '1 millisecond') + ON CONFLICT (topic_id, holder) DO UPDATE + SET position = least(f.position, excluded.position), until = excluded.until; +END $$; + +-- As in migration 2. The cap keeps the entries a driver took and works on. +CREATE OR REPLACE FUNCTION cap_log_v1(p_keep int) RETURNS bigint +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +DECLARE + v_partition int; + v_oldest bigint; + v_rows bigint; + deleted bigint := 0; +BEGIN + FOR v_partition IN 0..63 LOOP + SELECT n.id INTO v_oldest FROM notifications n WHERE n.partition = v_partition + ORDER BY n.id DESC OFFSET p_keep LIMIT 1; + IF FOUND THEN + -- An entry a driver took and still works on stays, or a quiet stream that shares the + -- partition with a busy one loses its only entry. One taken longer than a driver's + -- default `claim_idle` counts as left behind. + DELETE FROM notifications n WHERE n.partition = v_partition AND n.id <= v_oldest + AND (n.taken_by IS NULL OR n.taken_at < now() - interval '1 minute'); + GET DIAGNOSTICS v_rows = ROW_COUNT; + deleted := deleted + v_rows; + END IF; + END LOOP; + RETURN deleted; +END $$; + +-- As in migration 3. The hold and `archived` move forward only. +CREATE OR REPLACE FUNCTION advance_archive_v1(p_namespace text, p_kind smallint, p_owner text, + p_run text, p_activity text, p_topic text, p_position bigint, p_lease_ms bigint) RETURNS void +LANGUAGE plpgsql SET search_path FROM CURRENT AS $$ +DECLARE + v_topic bigint; + v_archived bigint; + v_closed boolean; +BEGIN + -- The key share lock keeps the janitor from deleting the topic under the insert. + SELECT t.topic_id, t.archived, t.closed INTO v_topic, v_archived, v_closed FROM chains c + JOIN topics t ON t.chain_id = c.chain_id AND t.topic = p_topic + WHERE (c.namespace, c.owner_kind, c.owner_id, c.first_run_id, c.activity_id) + = (p_namespace, p_kind, p_owner, p_run, p_activity) + FOR KEY SHARE OF t; + IF NOT FOUND THEN + RETURN; + END IF; + -- Two drivers can archive from the same start. The slower one's advance must not move the + -- hold back, nor hold a closed topic again after the archive released it. + IF p_position < v_archived OR (p_position = v_archived AND v_closed) THEN + RETURN; + END IF; + INSERT INTO floors AS f + VALUES (v_topic, 'archive', p_position, now() + p_lease_ms * interval '1 millisecond') + ON CONFLICT (topic_id, holder) DO UPDATE + SET position = excluded.position, until = excluded.until; + UPDATE topics t SET archived = p_position, + archive_pending = t.archive_pending AND t.last_position IS DISTINCT FROM p_position + WHERE t.topic_id = v_topic; +END $$; diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index f2d817d238..d30ddc0764 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -2145,6 +2145,21 @@ mod cases { assert_eq!(stored(&page.records[2]).link, None); } + pub(crate) async fn a_workflow_activity_keeps_a_topic_whose_first_batch_was_lost( + case: &dyn Case, + ) { + let chain = case.chain(); + let mut producer = Producer::new(case, &chain, OUT, "p", 1); + producer.sequence = 3; + producer.append(&["3"]).await; + let written = case.store().written_topics(&chain).await.unwrap(); + if chain.owner_kind == StreamOwnerKind::WorkflowActivity as i32 { + assert_eq!(written, [OUT]); + } else { + assert!(written.is_empty()); + } + } + pub(crate) async fn a_workflow_activity_keeps_the_topics_it_wrote(case: &dyn Case) { let store = case.store(); let chain = case.chain(); @@ -2882,6 +2897,7 @@ macro_rules! store_cases { a_close_with_no_outcome_is_the_first_close_too, a_chain_close_keeps_no_outcome, a_workflow_activity_keeps_the_topics_it_wrote, + a_workflow_activity_keeps_a_topic_whose_first_batch_was_lost, a_link_record_reads_back_with_its_link, ); }; diff --git a/crates/streams/tests/postgres_store.rs b/crates/streams/tests/postgres_store.rs index 6593a7bf03..e60d84265b 100644 --- a/crates/streams/tests/postgres_store.rs +++ b/crates/streams/tests/postgres_store.rs @@ -681,3 +681,171 @@ async fn a_listener_that_lost_its_connection_wakes_every_waiter() { let waited = waiting.await.unwrap(); assert!(waited < Duration::from_secs(5), "{waited:?}"); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn held_topics_leave_the_janitor_room_for_the_others() { + let Some(url) = url() else { return }; + let options = options(&url); + let schema = options.schema.clone(); + let store = PostgresStore::connect(options).await.unwrap(); + let raw = raw(&url).await; + // More held topics than one pass takes, then one plain topic. + for topic in 0..600 { + let chain = chain(&format!("held-{topic}")); + store + .append(append(&chain, "out", "p", 1, 1)) + .await + .unwrap(); + store.hold_floor(&chain, "out", "r/1", "").await.unwrap(); + } + let plain = chain("plain"); + store + .append(append(&plain, "out", "p", 1, 1)) + .await + .unwrap(); + age(&raw, &schema, 8).await; + store.run_janitor().await.unwrap(); + store.run_janitor().await.unwrap(); + let left: i64 = raw + .query_one( + &format!( + "SELECT count(*) FROM {schema}.records r JOIN {schema}.topics t \ + ON t.topic_id = r.topic_id JOIN {schema}.chains c ON c.chain_id = t.chain_id \ + WHERE c.owner_id = 'plain'" + ), + &[], + ) + .await + .unwrap() + .get(0); + assert_eq!(left, 0, "the plain topic was trimmed"); +} + +#[tokio::test] +async fn a_topic_refuses_a_floor_holder_past_the_cap() { + let Some(url) = url() else { return }; + let store = PostgresStore::connect(options(&url)).await.unwrap(); + let chain = chain("holders"); + for holder in 0..temporalio_streams::MAX_FLOOR_HOLDERS { + store + .hold_floor(&chain, "out", &format!("r/{holder}"), "") + .await + .unwrap(); + } + let error = store + .hold_floor(&chain, "out", "one-more/1", "") + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Refused, "{error}"); + store.hold_floor(&chain, "out", "r/0", "").await.unwrap(); +} + +#[tokio::test] +async fn two_archive_advances_out_of_order_keep_the_higher() { + let Some(url) = url() else { return }; + let mut options = options(&url); + options.notification_log = true; + options.archive_lease = Some(Duration::from_secs(60)); + let store = PostgresStore::connect(options).await.unwrap(); + let chain = chain("advances"); + let landed = store + .append(append(&chain, "out", "p", 1, 2)) + .await + .unwrap(); + store + .advance_archive(&chain, "out", &landed.last_position) + .await + .unwrap(); + store + .advance_archive(&chain, "out", &landed.first_position) + .await + .unwrap(); + assert_eq!( + store.archived(&chain, "out").await.unwrap(), + Some(landed.last_position) + ); +} + +#[tokio::test] +async fn an_append_rings_while_a_driver_holds_the_partitions_entries() { + let Some(url) = url() else { return }; + let mut options = options(&url); + options.notification_log = true; + let store = PostgresStore::connect(options).await.unwrap(); + let log = store.notification_log().unwrap(); + let chain = chain("rings"); + store + .append(append(&chain, "out", "p", 1, 1)) + .await + .unwrap(); + let rung = |doorbell| log.wait_doorbell(doorbell, "d", Duration::from_secs(2)); + let mut partition = None; + for doorbell in 0..log.doorbells() { + if let Some(&first) = rung(doorbell).await.unwrap().first() { + partition = Some((doorbell, first)); + } + } + let (doorbell, partition) = partition.expect("the first append rang"); + let taken = log + .take(partition, "d", Duration::from_secs(60), 10) + .await + .unwrap(); + assert_eq!(taken.len(), 1); + // The driver drains and goes back to its doorbell before this append commits. + store + .append(append(&chain, "out", "p", 2, 1)) + .await + .unwrap(); + let ids: Vec = taken.iter().map(|entry| entry.id.clone()).collect(); + log.ack(partition, &ids).await.unwrap(); + assert_eq!(rung(doorbell).await.unwrap(), [partition]); +} + +#[tokio::test] +async fn the_log_cap_keeps_an_entry_a_driver_works_on() { + let Some(url) = url() else { return }; + let mut options = options(&url); + options.notification_log = true; + let schema = options.schema.clone(); + let store = PostgresStore::connect(options).await.unwrap(); + let raw = raw(&url).await; + let log = store.notification_log().unwrap(); + let chain = chain("cap"); + store + .append(append(&chain, "out", "p", 1, 1)) + .await + .unwrap(); + let partition: i16 = raw + .query_one( + &format!("SELECT partition FROM {schema}.notifications"), + &[], + ) + .await + .unwrap() + .get(0); + let taken = log + .take(partition as u32, "d", Duration::from_secs(60), 10) + .await + .unwrap(); + for sequence in 2..=3 { + store + .append(append(&chain, "out", "p", sequence, 1)) + .await + .unwrap(); + } + raw.execute(&format!("SELECT {schema}.cap_log_v1(1)"), &[]) + .await + .unwrap(); + let left: Vec = raw + .query( + &format!("SELECT id FROM {schema}.notifications ORDER BY id"), + &[], + ) + .await + .unwrap() + .iter() + .map(|row| row.get(0)) + .collect(); + assert_eq!(left.len(), 2, "{left:?}"); + assert_eq!(left[0].to_string(), taken[0].id); +} From 48d4daf350cf85c966159b3291dfdcdf0118adc4 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 06:04:07 -0700 Subject: [PATCH 3/3] Lapsed the archive hold per store in the lost-records case. A PostgreSQL release keeps the hold for its grace, so the case ended it the way each store's lease ends. --- crates/streams/tests/conformance.rs | 41 ++++++++++++++++++++++++++--- 1 file changed, 37 insertions(+), 4 deletions(-) diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index d30ddc0764..bc73496b9d 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -131,6 +131,14 @@ impl Case for Notifying { #[async_trait::async_trait] trait Forgets: Case { async fn forget(&self, chain: &ChainId, topic: &str); + + /// Ends the archive's hold on a topic at once, as a lapsed lease does. + async fn lapse_archive_hold(&self, chain: &ChainId, topic: &str) { + self.store() + .release_floor(chain, topic, temporalio_streams::ARCHIVE_HOLDER) + .await + .unwrap(); + } } #[async_trait::async_trait] @@ -224,6 +232,9 @@ trait ArchiveCase: Case { /// Deletes one archive object, as a bucket lifecycle rule does. async fn lose(&self, name: &str); + /// Ends the archive's hold on a topic at once, as a lapsed lease does. + async fn lapse_archive_hold(&self, chain: &ChainId, topic: &str); + /// Archives what the hot store holds of a topic, in segments of at most `records` records. async fn archive_now(&self, chain: &ChainId, topic: &str, records: u32) { let limits = SegmentLimits { @@ -250,6 +261,10 @@ impl ArchiveCase for Tiered { self.inner.forget(chain, topic).await; } + async fn lapse_archive_hold(&self, chain: &ChainId, topic: &str) { + self.inner.lapse_archive_hold(chain, topic).await; + } + async fn lose(&self, name: &str) { match &self.root { Some(root) => std::fs::remove_file(root.join(name)).unwrap(), @@ -263,6 +278,10 @@ impl Forgets for Activity { async fn forget(&self, chain: &ChainId, topic: &str) { self.0.forget(chain, topic).await; } + + async fn lapse_archive_hold(&self, chain: &ChainId, topic: &str) { + self.0.lapse_archive_hold(chain, topic).await; + } } #[async_trait::async_trait] @@ -270,6 +289,10 @@ impl Forgets for WorkflowActivity { async fn forget(&self, chain: &ChainId, topic: &str) { self.0.forget(chain, topic).await; } + + async fn lapse_archive_hold(&self, chain: &ChainId, topic: &str) { + self.0.lapse_archive_hold(chain, topic).await; + } } async fn memory_holding_for_archive() -> Option { @@ -1084,6 +1107,19 @@ mod on_postgres { ) .await; } + + async fn lapse_archive_hold(&self, chain: &ChainId, topic: &str) { + self.on_topic( + &format!( + "WITH t AS ({TOPIC}) UPDATE {{s}}.floors f SET until = now() - interval '1 s' \ + FROM t WHERE f.topic_id = t.topic_id AND f.holder = 'archive'" + ), + chain, + topic, + &[], + ) + .await; + } } } @@ -2721,10 +2757,7 @@ mod archive_cases { let lost = producer.append(&["3"]).await; producer.append(&["4", "5"]).await; // The archive's hold lapses, and retention drops `3` before the archive takes it. - case.hot() - .release_floor(&chain, OUT, temporalio_streams::ARCHIVE_HOLDER) - .await - .unwrap(); + case.lapse_archive_hold(&chain, OUT).await; case.drop_oldest(&chain, OUT, 2).await; case.archive_now(&chain, OUT, 10).await; // Only the archive answers now.