fix(relay): reset sync for empty channel changes

Signed-off-by: Daniel Rouleau <drouleau@squareup.com>
This commit is contained in:
Daniel Rouleau
2026-10-21 17:46:26 -07:00
parent 6a52195bd9
commit c8d891ed0c
6 changed files with 105 additions and 8 deletions

View File

@@ -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?;
}

View File

@@ -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,

View File

@@ -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()));
}
}

View File

@@ -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;

View File

@@ -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.

View File

@@ -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"`.