diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 9b12d1533a..e984314d0a 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -915,9 +915,9 @@ pub async fn update_channel( } if policy_changed { sqlx::query( - "UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1 AND (EXISTS(SELECT 1 FROM artifact_heads WHERE community_id=$1 AND channel_id=$2) OR EXISTS(SELECT 1 FROM artifact_revisions WHERE community_id=$1 AND source_channel_id=$2))", + "UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1", ) - .bind(community_id.as_uuid()).bind(channel_id) + .bind(community_id.as_uuid()) .execute(tx.conn()) .await?; } @@ -1119,9 +1119,9 @@ pub async fn soft_delete_channel( if result.rows_affected() > 0 { sqlx::query( - "UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1 AND (EXISTS(SELECT 1 FROM artifact_heads WHERE community_id=$1 AND channel_id=$2) OR EXISTS(SELECT 1 FROM artifact_revisions WHERE community_id=$1 AND source_channel_id=$2))", + "UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1", ) - .bind(community_id.as_uuid()).bind(channel_id) + .bind(community_id.as_uuid()) .execute(tx.conn()) .await?; } diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs index 03dcdef264..0855cd1c3a 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs @@ -126,7 +126,7 @@ async fn sync_rejoin_replays_old_heads_without_resetting_on_task_edits() { #[tokio::test] #[ignore = "requires Postgres"] -async fn sync_ignores_unrelated_ttl_and_unchanged_visibility() { +async fn sync_ignores_unchanged_visibility_and_resets_on_ttl_changes() { let _guard = SERIAL.lock().await; let f = fixture().await; accept(&f, &f.revision(Uuid::new_v4(), "create", None)).await; @@ -143,6 +143,8 @@ async fn sync_ignores_unrelated_ttl_and_unchanged_visibility() { ) .await .unwrap(); + let renamed = sync(&f, Some(&first["checkpoint"])).await; + assert!(renamed["checkpoint"].is_string()); buzz_db::channel::update_channel( &f.pool, f.community, @@ -154,7 +156,15 @@ async fn sync_ignores_unrelated_ttl_and_unchanged_visibility() { ) .await .unwrap(); - let next = sync(&f, Some(&first["checkpoint"])).await; + let (status, reset) = f + .request( + "/query", + json!([{"artifact":"sync","query":query(),"checkpoint":renamed["checkpoint"]}]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(reset["reason"], "policy_changed"); + let next = sync(&f, None).await; assert!(next["checkpoint"].is_string()); accept( &f, diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_policy_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_policy_postgres_tests.rs new file mode 100644 index 0000000000..954a313cd6 --- /dev/null +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_policy_postgres_tests.rs @@ -0,0 +1,79 @@ +use super::*; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_resets_for_first_artifact_during_channel_transition() { + let _guard = SERIAL.lock().await; + for delete_channel in [false, true] { + let f = fixture().await; + let mut blocker = f.pool.begin().await.unwrap(); + let blocker_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *blocker) + .await + .unwrap(); + sqlx::query("SELECT id FROM communities WHERE id=$1 FOR NO KEY UPDATE") + .bind(f.community.as_uuid()) + .fetch_one(&mut *blocker) + .await + .unwrap(); + let pool = f.pool.clone(); + let community = f.community; + let home = f.home; + let transition = tokio::spawn(async move { + if delete_channel { + assert!( + buzz_db::channel::soft_delete_channel(&pool, community, home) + .await + .unwrap() + ); + } else { + buzz_db::channel::update_channel( + &pool, + community, + home, + buzz_db::channel::ChannelUpdate { + visibility: Some("private".into()), + ..Default::default() + }, + ) + .await + .unwrap(); + } + }); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + assert!(!transition.is_finished(), "an empty channel must still advance the recovery marker"); + let blocked: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE $1=ANY(pg_blocking_pids(pid)) AND query LIKE 'UPDATE communities SET artifact_policy_modified_at=%')") + .bind(blocker_pid).fetch_one(&f.pool).await.unwrap(); + if blocked { break; } + tokio::task::yield_now().await; + } + }).await.expect("channel transition reaches the recovery marker"); + let id = Uuid::new_v4(); + let event = f.revision(id, "create", None); + tokio::time::timeout(std::time::Duration::from_secs(2), accept(&f, &event)) + .await + .expect("first artifact commits while the channel transition is uncommitted"); + let (status, first) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"sync","query":query()}]), + ) + .await; + assert!(status.is_success(), "{status}: {first}"); + assert_eq!(first["changes"][0]["event"]["id"], event.id.to_hex()); + blocker.commit().await.unwrap(); + transition.await.unwrap(); + let (status, next) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"sync","query":query(),"checkpoint":first["checkpoint"]}]), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(next["reason"], "policy_changed"); + assert!(!next.to_string().contains(&f.home.to_string())); + } +} diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs index 751ca3cd04..3a776dc158 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs @@ -2,6 +2,8 @@ use super::*; use axum::http::StatusCode; #[path = "sync_access_postgres_tests.rs"] mod access_postgres_tests; +#[path = "sync_policy_postgres_tests.rs"] +mod policy_postgres_tests; #[path = "sync_scale_postgres_tests.rs"] mod scale_postgres_tests; diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md index 719f1f46ca..1b68c7e9b2 100644 --- a/docs/artifact-sync-deployment.md +++ b/docs/artifact-sync-deployment.md @@ -21,6 +21,7 @@ CREATE INDEX CONCURRENTLY idx_artifact_revisions_artifact ON public.artifact_revisions (community_id, artifact_id); ``` +Apply `0068_artifact_revision_lookup.sql` in a transaction after that prebuild. Then apply `0069_artifact_sync_timestamps.sql` in a transaction with its lock and statement budgets. The migration is idempotent for this prebuild step. Then build these indexes outside a transaction: @@ -43,6 +44,11 @@ database can build the indexes under the migration's short budgets. Migration `0068` also verifies the exact definition and all three flags. Remove an invalid index with `DROP INDEX CONCURRENTLY` before a new build. +After all index builds succeed, run `buzz-admin migrate` with the target +`DATABASE_URL`. It verifies and records `0068`, reapplies the idempotent `0069`, +then verifies and records `0070`. Direct SQL application does not record the +migration ledger. + ## Enable and rollback Deploy the timestamp writers to every relay instance with the gate off. diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md index 1aee61a19b..f91aa0dd67 100644 --- a/docs/nips/NIP-AR.md +++ b/docs/nips/NIP-AR.md @@ -195,8 +195,8 @@ or remove. An empty delta does not confirm an accepted edit. ### Access recovery and bounds Relevant channel access changes atomically advance one community policy -stamp. The check includes channels with current heads or retained move-source -records. Ordinary artifact writes do not update this stamp. A different stamp +stamp, including transitions on empty channels. Ordinary artifact writes do +not update this stamp. A different stamp returns HTTP 409 `sync_reset` with `reason: "policy_changed"`, without naming private channels. A checkpoint older than 30 days returns the same code with `reason: "checkpoint_expired"`.