diff --git a/crates/buzz-db/src/store/artifact_query.rs b/crates/buzz-db/src/store/artifact_query.rs index e1de756c52..80fe6b47a8 100644 --- a/crates/buzz-db/src/store/artifact_query.rs +++ b/crates/buzz-db/src/store/artifact_query.rs @@ -55,7 +55,7 @@ impl Db { } } -/// Apply the same visibility and tag rules to pages and snapshot capture. +/// Own the visibility and tag rules for artifact queries and current-ID reads. pub(super) fn predicates( qb: &mut QueryBuilder, community: CommunityId, diff --git a/crates/buzz-db/src/store/artifact_sync.rs b/crates/buzz-db/src/store/artifact_sync.rs index 18b9323276..127910c228 100644 --- a/crates/buzz-db/src/store/artifact_sync.rs +++ b/crates/buzz-db/src/store/artifact_sync.rs @@ -167,6 +167,8 @@ impl Db { } } +// Keep candidate access rules aligned with artifact_query::predicates, which owns +// current-state visibility. This query also reads retained move and membership rows. const CANDIDATES: &str = r#" WITH readable AS NOT MATERIALIZED ( SELECT c.id FROM channels c diff --git a/crates/buzz-relay/src/api/artifact.rs b/crates/buzz-relay/src/api/artifact.rs index 9a60763a88..c41ce5d15d 100644 --- a/crates/buzz-relay/src/api/artifact.rs +++ b/crates/buzz-relay/src/api/artifact.rs @@ -47,9 +47,12 @@ async fn execute( "artifact queries require exactly one filter", )); } - if filters[0].get("artifact").and_then(Value::as_str) == Some("sync") { + if filters[0].get("artifact").and_then(Value::as_str) == Some("incremental") { if count { - return Err(api_error(StatusCode::BAD_REQUEST, "sync requires /query")); + return Err(api_error( + StatusCode::BAD_REQUEST, + "incremental requires /query", + )); } return super::artifact_sync::query(state, tenant, reader, &filters[0]).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 af52ff02e1..58f63851bf 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 @@ -20,7 +20,7 @@ async fn sync_redaction_and_query_exit_return_removals() { accept(&f, &first).await; let scope = json!({"artifact":"current","#type":["buzz.task"],"#project":["alpha"]}); let (_, page) = f - .request("/query", json!([{"artifact":"sync","query":scope}])) + .request("/query", json!([{"artifact":"incremental","query":scope}])) .await; assert_eq!(page["changes"][0]["type"], "upsert"); let changed = f.revision(id, "update", Some(&first)); @@ -28,7 +28,7 @@ async fn sync_redaction_and_query_exit_return_removals() { let (status, delta) = f .request( "/query", - json!([{"artifact":"sync","query":scope,"checkpoint":page["checkpoint"]}]), + json!([{"artifact":"incremental","query":scope,"checkpoint":page["checkpoint"]}]), ) .await; assert!(status.is_success(), "{status}: {delta}"); @@ -80,7 +80,7 @@ async fn sync_rejoin_replays_old_heads_without_resetting_on_task_edits() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query()}]), + json!([{"artifact":"incremental","query":query()}]), ) .await; buzz_db::channel_members::remove_member( @@ -106,7 +106,7 @@ async fn sync_rejoin_replays_old_heads_without_resetting_on_task_edits() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":page["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":page["checkpoint"]}]), ) .await; assert!(status.is_success(), "{status}: {delta}"); @@ -117,7 +117,7 @@ async fn sync_rejoin_replays_old_heads_without_resetting_on_task_edits() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":delta["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":delta["checkpoint"]}]), ) .await; assert!(status.is_success(), "{status}: {delta}"); @@ -159,7 +159,7 @@ async fn sync_ignores_unchanged_visibility_and_resets_on_ttl_changes() { let (status, reset) = f .request( "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":renamed["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":renamed["checkpoint"]}]), ) .await; assert_eq!(status, StatusCode::CONFLICT); @@ -172,10 +172,13 @@ async fn sync_ignores_unchanged_visibility_and_resets_on_ttl_changes() { ) .await; let (status, body) = f - .request("/query", json!([{"artifact":"sync","query":query()}])) + .request( + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) .await; assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY); - assert_eq!(body["code"], "sync_unsupported_scope"); + assert_eq!(body["code"], "incremental_unsupported_scope"); } #[tokio::test] @@ -200,7 +203,7 @@ async fn sync_shrinks_large_pages_and_current_batches_fail_explicitly() { let (status, last) = f .request( "/query", - json!([{"artifact":"sync","query":query(),"cursor":first["nextCursor"]}]), + json!([{"artifact":"incremental","query":query(),"cursor":first["nextCursor"]}]), ) .await; assert!(status.is_success(), "{status}: {last}"); @@ -209,11 +212,11 @@ async fn sync_shrinks_large_pages_and_current_batches_fail_explicitly() { let (status, body) = f .request( "/query", - json!([{"artifact":"sync","action":"current","query":query(),"artifactIds":ids}]), + json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":ids}]), ) .await; assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE); - assert_eq!(body["code"], "sync_limit"); + assert_eq!(body["code"], "incremental_limit"); } #[tokio::test] @@ -222,8 +225,10 @@ async fn sync_gate_controls_http_reads_and_capability() { use tower::ServiceExt; let _guard = SERIAL.lock().await; let mut f = fixture().await; + let mut websocket_sync = Value::Null; for enabled in [false, true] { - Arc::make_mut(&mut Arc::make_mut(&mut f.state).config).artifact_sync_enabled = enabled; + Arc::make_mut(&mut Arc::make_mut(&mut f.state).config).artifact_incremental_enabled = + enabled; let response = crate::router::build_router(f.state.clone()) .oneshot( axum::http::Request::builder() @@ -240,17 +245,22 @@ async fn sync_gate_controls_http_reads_and_capability() { .unwrap(); let info: Value = serde_json::from_slice(&bytes).unwrap(); if enabled { + assert_eq!(info["artifacts"]["sync"], websocket_sync); assert_eq!( - info["artifacts"]["sync"], + info["artifacts"]["incremental"], json!({"version":1,"overlapSeconds":120, "pageRows":100,"currentRows":100,"responseBytes":8388608, "permanentChannelsOnly":true,"checkpointLifetimeSeconds":2592000}) ); } else { - assert!(info["artifacts"].get("sync").is_none()); + websocket_sync = info["artifacts"]["sync"].clone(); + assert!(info["artifacts"].get("incremental").is_none()); } let (status, _) = f - .request("/query", json!([{"artifact":"sync","query":query()}])) + .request( + "/query", + json!([{"artifact":"incremental","query":query()}]), + ) .await; assert_eq!( status, @@ -265,7 +275,7 @@ async fn sync_gate_controls_http_reads_and_capability() { #[tokio::test] #[ignore = "requires Postgres"] -async fn sync_resets_when_a_move_source_loses_access() { +async fn incremental_resets_when_a_move_source_loses_access() { let _guard = SERIAL.lock().await; for delete_channel in [false, true] { let f = fixture().await; @@ -276,7 +286,7 @@ async fn sync_resets_when_a_move_source_loses_access() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query()}]), + json!([{"artifact":"incremental","query":query()}]), ) .await; let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&first)); @@ -313,7 +323,7 @@ async fn sync_resets_when_a_move_source_loses_access() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":read["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":read["checkpoint"]}]), ) .await; assert_eq!(status, StatusCode::CONFLICT); @@ -323,7 +333,7 @@ async fn sync_resets_when_a_move_source_loses_access() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","action":"current","query":query(),"artifactIds":[id]}]), + json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":[id]}]), ) .await; assert!(status.is_success()); @@ -371,14 +381,14 @@ async fn sync_progress_replays_queries_with_escaped_values() { let f = fixture().await; let scope = json!({"artifact":"current","#type":["buzz.task"],"#project":vec!["\u{1}".repeat(4096);15]}); let (status, first) = f - .request("/query", json!([{"artifact":"sync","query":scope}])) + .request("/query", json!([{"artifact":"incremental","query":scope}])) .await; assert!(status.is_success(), "{status}: {first}"); assert!(first["checkpoint"].as_str().unwrap().len() < 200_000); let (status, next) = f .request( "/query", - json!([{"artifact":"sync","query":scope,"checkpoint":first["checkpoint"]}]), + json!([{"artifact":"incremental","query":scope,"checkpoint":first["checkpoint"]}]), ) .await; assert!(status.is_success(), "{status}: {next}"); @@ -398,7 +408,7 @@ async fn sync_current_intersects_known_ids_with_query_scope() { let (status, body) = f .request( "/query", - json!([{"artifact":"sync","action":"current", + json!([{"artifact":"incremental","action":"current", "query":{"artifact":"current","#type":["buzz.task"],"#d":[included]}, "artifactIds":[included,excluded]}]), ) diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs index 4f2b855e57..ddc71968f6 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs @@ -9,7 +9,7 @@ async fn sync_move_cursor_hides_private_and_redacted_revision_ids() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query()}]), + json!([{"artifact":"incremental","query":query()}]), ) .await; let mut current = std::collections::HashMap::new(); @@ -45,7 +45,7 @@ async fn sync_move_cursor_hides_private_and_redacted_revision_ids() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":empty["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":empty["checkpoint"]}]), ) .await; assert!(status.is_success(), "{status}: {first}"); @@ -66,7 +66,7 @@ async fn sync_move_cursor_hides_private_and_redacted_revision_ids() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"cursor":cursor}]), + json!([{"artifact":"incremental","query":query(),"cursor":cursor}]), ) .await; assert!(status.is_success(), "{status}: {last}"); 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 index 954a313cd6..c3cc3d9f1d 100644 --- 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 @@ -2,7 +2,7 @@ use super::*; #[tokio::test] #[ignore = "requires Postgres"] -async fn sync_resets_for_first_artifact_during_channel_transition() { +async fn incremental_resets_for_first_artifact_during_channel_transition() { let _guard = SERIAL.lock().await; for delete_channel in [false, true] { let f = fixture().await; @@ -58,7 +58,7 @@ async fn sync_resets_for_first_artifact_during_channel_transition() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query()}]), + json!([{"artifact":"incremental","query":query()}]), ) .await; assert!(status.is_success(), "{status}: {first}"); @@ -69,7 +69,7 @@ async fn sync_resets_for_first_artifact_during_channel_transition() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":first["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":first["checkpoint"]}]), ) .await; assert_eq!(status, StatusCode::CONFLICT); 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 aaac2be864..a5025b8dca 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 @@ -12,7 +12,7 @@ mod scale_postgres_tests; async fn fixture() -> Fixture { let mut f = Fixture::new().await; let mut state = (*f.state).clone(); - Arc::make_mut(&mut state.config).artifact_sync_enabled = true; + Arc::make_mut(&mut state.config).artifact_incremental_enabled = true; f.state = Arc::new(state); f } @@ -22,7 +22,7 @@ fn query() -> Value { json!({"artifact":"current","#type":["buzz.task"]}) } async fn sync(f: &Fixture, checkpoint: Option<&Value>) -> Value { - let mut request = json!({"artifact":"sync","query":query()}); + let mut request = json!({"artifact":"incremental","query":query()}); if let Some(checkpoint) = checkpoint { request["checkpoint"] = checkpoint.clone(); } @@ -103,7 +103,7 @@ async fn sync_pages_equal_timestamps_and_replays_the_same_cursor() { let first = sync(&f, None).await; assert_eq!(first["changes"].as_array().unwrap().len(), 100); assert!(first["checkpoint"].is_null()); - let request = json!([{"artifact":"sync","query":query(),"cursor":first["nextCursor"]}]); + let request = json!([{"artifact":"incremental","query":query(),"cursor":first["nextCursor"]}]); let (status, last) = f.request("/query", request.clone()).await; assert!(status.is_success(), "{status}: {last}"); assert_eq!(last["changes"].as_array().unwrap().len(), 3); @@ -130,7 +130,7 @@ async fn sync_current_has_uniform_unavailable_and_rejects_duplicate_ids() { let event = f.revision_as(&f.owner, f.private, id, "create", None); accept(&f, &event).await; let missing = Uuid::new_v4(); - let body = json!([{"artifact":"sync","action":"current","query":query(),"artifactIds":[id,missing,"local-only"]}]); + let body = json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":[id,missing,"local-only"]}]); let (status, body) = f.request_as(&f.peer, "/query", body).await; assert!(status.is_success(), "{status}: {body}"); assert_eq!( @@ -142,14 +142,14 @@ async fn sync_current_has_uniform_unavailable_and_rejects_duplicate_ids() { let (status, _) = f .request( "/query", - json!([{"artifact":"sync","action":"current","query":query(),"artifactIds":[id,id]}]), + json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":[id,id]}]), ) .await; assert_eq!(status, StatusCode::BAD_REQUEST); let (status, body) = f .request( "/query", - json!([{"artifact":"sync","action":"current","query":query(),"artifactIds":[id]}]), + json!([{"artifact":"incremental","action":"current","query":query(),"artifactIds":[id]}]), ) .await; assert!(status.is_success(), "{status}: {body}"); @@ -173,7 +173,7 @@ async fn sync_grant_finds_old_heads_and_removal_names_only_own_membership() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query()}]), + json!([{"artifact":"incremental","query":query()}]), ) .await; assert_eq!(first["changes"], json!([])); @@ -191,7 +191,7 @@ async fn sync_grant_finds_old_heads_and_removal_names_only_own_membership() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":first["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":first["checkpoint"]}]), ) .await; assert!( @@ -215,7 +215,7 @@ async fn sync_grant_finds_old_heads_and_removal_names_only_own_membership() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":granted["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":granted["checkpoint"]}]), ) .await; assert_eq!( @@ -236,7 +236,7 @@ async fn sync_move_to_hidden_destination_removes_the_source_view() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query()}]), + json!([{"artifact":"incremental","query":query()}]), ) .await; let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&first)); @@ -245,7 +245,7 @@ async fn sync_move_to_hidden_destination_removes_the_source_view() { .request_as( &f.peer, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":read["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":read["checkpoint"]}]), ) .await; assert_eq!( @@ -279,13 +279,13 @@ async fn sync_policy_changes_reset_without_hidden_channel_ids() { let (status, body) = f .request( "/query", - json!([{"artifact":"sync","query":query(),"cursor":first["nextCursor"]}]), + json!([{"artifact":"incremental","query":query(),"cursor":first["nextCursor"]}]), ) .await; assert_eq!(status, StatusCode::CONFLICT); assert_eq!( body, - json!({"code":"sync_reset","error":"sync_reset","reason":"policy_changed"}) + json!({"code":"incremental_reset","error":"incremental_reset","reason":"policy_changed"}) ); } @@ -306,7 +306,7 @@ async fn sync_progress_binds_reader_query_and_community() { .request_as( reader, "/query", - json!([{"artifact":"sync","query":scope,"checkpoint":first["checkpoint"]}]), + json!([{"artifact":"incremental","query":scope,"checkpoint":first["checkpoint"]}]), ) .await; assert_eq!(status, StatusCode::BAD_REQUEST); @@ -317,7 +317,7 @@ async fn sync_progress_binds_reader_query_and_community() { .request_as( &f.owner, "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":first["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":first["checkpoint"]}]), ) .await; assert_eq!(status, StatusCode::BAD_REQUEST); @@ -329,7 +329,7 @@ async fn sync_progress_binds_reader_query_and_community() { let (status, body) = f .request( "/query", - json!([{"artifact":"sync","query":query(),"checkpoint":expired}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":expired}]), ) .await; assert_eq!(status, StatusCode::CONFLICT); diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_scale_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_scale_postgres_tests.rs index 130b1ab43f..cf23203eaa 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_scale_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_scale_postgres_tests.rs @@ -46,7 +46,7 @@ async fn sync_downloads_ten_thousand_tasks_with_sixty_thousand_memberships() { let mut other = (*bridge_handler_test_state().await.unwrap()).clone(); let config = Arc::make_mut(&mut other.config); config.require_auth_token = true; - config.artifact_sync_enabled = true; + config.artifact_incremental_enabled = true; other.relay_keypair = f.state.relay_keypair.clone(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("http://{}/query", listener.local_addr().unwrap()); @@ -73,7 +73,7 @@ async fn sync_downloads_ten_thousand_tasks_with_sixty_thousand_memberships() { &f, &client, &url, - json!([{"artifact":"sync","query":query(),"cursor":page["nextCursor"]}]), + json!([{"artifact":"incremental","query":query(),"cursor":page["nextCursor"]}]), ) .await; assert!(status.is_success(), "{status}: {next}"); @@ -87,7 +87,7 @@ async fn sync_downloads_ten_thousand_tasks_with_sixty_thousand_memberships() { &f, &client, &url, - json!([{"artifact":"sync","query":query(),"checkpoint":page["checkpoint"]}]), + json!([{"artifact":"incremental","query":query(),"checkpoint":page["checkpoint"]}]), ) .await; assert!(status.is_success(), "{status}: {delta}"); diff --git a/crates/buzz-relay/src/api/artifact_sync.rs b/crates/buzz-relay/src/api/artifact_sync.rs index 7bd50dfb3b..23373eede9 100644 --- a/crates/buzz-relay/src/api/artifact_sync.rs +++ b/crates/buzz-relay/src/api/artifact_sync.rs @@ -1,4 +1,4 @@ -//! Timestamp sync with signed progress and bounded current-identity reads. +//! Incremental reads with signed progress and bounded current-identity reads. use crate::state::AppState; use axum::{http::StatusCode, Json}; use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; @@ -53,28 +53,29 @@ fn error(status: StatusCode, code: &str) -> (StatusCode, Json) { (status, Json(json!({"error":code,"code":code}))) } fn invalid() -> (StatusCode, Json) { - error(StatusCode::BAD_REQUEST, "sync_invalid") + error(StatusCode::BAD_REQUEST, "incremental_invalid") } fn unavailable() -> (StatusCode, Json) { - error(StatusCode::SERVICE_UNAVAILABLE, "sync_unavailable") + error(StatusCode::SERVICE_UNAVAILABLE, "incremental_unavailable") } fn limit() -> (StatusCode, Json) { - error(StatusCode::PAYLOAD_TOO_LARGE, "sync_limit") + error(StatusCode::PAYLOAD_TOO_LARGE, "incremental_limit") } fn reset(reason: &str) -> (StatusCode, Json) { ( StatusCode::CONFLICT, - Json(json!({"error":"sync_reset","code":"sync_reset","reason":reason})), + Json(json!({"error":"incremental_reset","code":"incremental_reset","reason":reason})), ) } fn db_error(e: buzz_db::DbError) -> (StatusCode, Json) { match e { buzz_db::DbError::ArtifactSyncLimit => limit(), - buzz_db::DbError::ArtifactSyncUnsupportedScope => { - error(StatusCode::UNPROCESSABLE_ENTITY, "sync_unsupported_scope") - } + buzz_db::DbError::ArtifactSyncUnsupportedScope => error( + StatusCode::UNPROCESSABLE_ENTITY, + "incremental_unsupported_scope", + ), error => { - tracing::warn!(%error,"artifact sync database read failed"); + tracing::warn!(%error,"incremental artifact read failed"); unavailable() } } @@ -89,7 +90,7 @@ pub(crate) fn capability() -> Value { fn mac(state: &AppState) -> Result, (StatusCode, Json)> { let mut digest = Sha256::new(); - digest.update(b"buzz-artifact-sync-v1\0"); + digest.update(b"buzz-artifact-incremental-v1\0"); digest.update(state.relay_keypair.secret_key().as_secret_bytes()); Hmac::::new_from_slice(&digest.finalize()).map_err(|_| unavailable()) } @@ -120,7 +121,7 @@ fn decode(state: &AppState, value: &str) -> Result, @@ -128,7 +129,7 @@ pub(super) async fn query( reader: &nostr::PublicKey, filter: &Value, ) -> Response { - if !state.config.artifact_sync_enabled { + if !state.config.artifact_incremental_enabled { return Err(unavailable()); } let _permit = WORK.try_acquire().map_err(|_| unavailable())?; @@ -147,7 +148,7 @@ async fn execute( filter: &Value, ) -> Response { let request: Request = serde_json::from_value(filter.clone()).map_err(|_| invalid())?; - if request.artifact != "sync" + if request.artifact != "incremental" || request.query.get("limit").is_some() || request.query.get("offset").is_some() { diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 7ad4ee1a02..b9ec10612c 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -117,7 +117,7 @@ pub const MAX_DRAIN_JITTER_MS: u64 = 20_000; #[derive(Debug, Clone)] pub struct Config { /// Enable incremental artifact reads after every serving writer is upgraded. - pub artifact_sync_enabled: bool, + pub artifact_incremental_enabled: bool, /// Address the relay HTTP/WebSocket server binds to. pub bind_addr: SocketAddr, /// Postgres database connection URL. @@ -774,7 +774,7 @@ impl Config { let require_relay_membership = std::env::var("BUZZ_REQUIRE_RELAY_MEMBERSHIP") .map(|v| v == "true" || v == "1") .unwrap_or(false); - let artifact_sync_enabled = std::env::var("BUZZ_ARTIFACT_SYNC_ENABLED") + let artifact_incremental_enabled = std::env::var("BUZZ_ARTIFACT_INCREMENTAL_ENABLED") .map(|v| v == "true" || v == "1") .unwrap_or(false); @@ -1432,7 +1432,7 @@ impl Config { partition_manager_advance_enabled, pubkey_allowlist_enabled, require_relay_membership, - artifact_sync_enabled, + artifact_incremental_enabled, huddle_audio_available, mesh, mesh_demo_echo, diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index 0146985580..b7fabb0828 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -387,8 +387,8 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st admin_api.as_deref(), state.config.klipy.as_ref().map(|_| "klipy"), ); - if state.config.artifact_sync_enabled { - info.artifacts["sync"] = crate::api::artifact_sync::capability(); + if state.config.artifact_incremental_enabled { + info.artifacts["incremental"] = crate::api::artifact_sync::capability(); } if let Ok(tenant) = crate::tenant::bind_community(&state.db, raw_host).await { if state.config.buzz_v1_enabled { diff --git a/docs/artifact-sync-deployment.md b/docs/artifact-sync-deployment.md index 12b4887314..2ed5876d69 100644 --- a/docs/artifact-sync-deployment.md +++ b/docs/artifact-sync-deployment.md @@ -1,7 +1,7 @@ -# Artifact sync deployment +# Incremental artifact reads deployment -Incremental artifact sync uses a 120-second overlap and database modification -timestamps. `BUZZ_ARTIFACT_SYNC_ENABLED` is off by default. Enable it only +Incremental artifact reads use a 120-second overlap and database modification +timestamps. `BUZZ_ARTIFACT_INCREMENTAL_ENABLED` is off by default. Enable it only after the schema, indexes, and every serving writer are upgraded. ## Schema and indexes @@ -54,8 +54,8 @@ migration ledger. Deploy the timestamp writers to every relay instance with the gate off. Drain the old writers. All instances must use the same configured relay signing key, because that key authenticates the progress tokens. Then enable -`BUZZ_ARTIFACT_SYNC_ENABLED=true` on the serving fleet. NIP-11 advertises -`artifacts.sync` only when the gate is enabled. The endpoint refuses sync +`BUZZ_ARTIFACT_INCREMENTAL_ENABLED=true` on the serving fleet. NIP-11 advertises +`artifacts.incremental` only when the gate is enabled. The endpoint refuses incremental requests while the gate is off. Before a rollback to an old writer, disable the gate on every serving diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md index f91aa0dd67..51fabf1299 100644 --- a/docs/nips/NIP-AR.md +++ b/docs/nips/NIP-AR.md @@ -107,16 +107,20 @@ Clients validate content before interpreting it as a known type. Unknown types o Artifact text is untrusted. A type name, relationship tag, or payload cannot authorize a privileged action. Moving, deleting, or revoking access cannot erase copies already downloaded. -## Incremental sync +## Incremental reads -The relay advertises `artifacts.sync` when incremental reads are enabled. +Incremental reads maintain an eventually consistent local store. They do not prove a complete set at one database snapshot. Absence from a page never proves deletion or loss of access. + +This HTTP protocol uses `artifacts.incremental`. The WebSocket Sync subscription uses a separate `artifacts.sync` capability and defines its own complete-set guarantee. Clients select each protocol through its own capability. + +The relay advertises `artifacts.incremental` when incremental reads are enabled. Each signed `POST /query` request contains one filter. The query uses the same community, relay admission, channel access, and exact tag rules as an ordinary artifact read. The existing `current` and `history` array responses keep their format. ```json -[{"artifact":"sync","query":{"artifact":"current","#type":["buzz.task"]}}] +[{"artifact":"incremental","query":{"artifact":"current","#type":["buzz.task"]}}] ``` The initial read pages through current identities. A later read supplies the @@ -176,14 +180,14 @@ current revalidation result without destination metadata. ### Current reads and write confirmation ```json -[{"artifact":"sync","action":"current","query":{"artifact":"current","#type":["buzz.task"]},"artifactIds":["22222222-2222-4222-8222-222222222222"]}] +[{"artifact":"incremental","action":"current","query":{"artifact":"current","#type":["buzz.task"]},"artifactIds":["22222222-2222-4222-8222-222222222222"]}] ``` The response has `version`, database `startedAt`, and `changes`. It contains exactly one upsert or remove for each requested ID, in request order. The request permits 1–100 unique nonempty strings, each at most 256 UTF-8 bytes. A string that is not a canonical artifact UUID returns remove. -Duplicates and invalid bounds fail with `sync_invalid`. The server does not +Duplicates and invalid bounds fail with `incremental_invalid`. The server does not truncate the request. Clients serialize a full delta pass and its commits before a current-ID read. @@ -197,7 +201,7 @@ or remove. An empty delta does not confirm an accepted edit. Relevant channel access changes atomically advance one community policy 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 +returns HTTP 409 `incremental_reset` with `reason: "policy_changed"`, without naming private channels. A checkpoint older than 30 days returns the same code with `reason: "checkpoint_expired"`. @@ -219,18 +223,18 @@ NIP-11 advertises: ``` Pages shrink to the response byte bound. A single oversized result returns -HTTP 413 `sync_limit`. A current-ID batch can return that error; the client +HTTP 413 `incremental_limit`. A current-ID batch can return that error; the client can split the batch. No page or batch silently truncates its result. The server permits four concurrent operations per process, a two-second SQL statement budget, and an eight-second operation budget. There is no identity-wide membership count limit on this path. Only matching readable artifacts in permanent channels are supported. -A matching TTL artifact returns HTTP 422 `sync_unsupported_scope`. An -unrelated TTL channel does not block task sync. HTTP 400 `sync_invalid` -rejects invalid requests or tokens. HTTP 503 `sync_unavailable` reports a +A matching TTL artifact returns HTTP 422 `incremental_unsupported_scope`. An +unrelated TTL channel does not block incremental task reads. HTTP 400 `incremental_invalid` +rejects invalid requests or tokens. HTTP 503 `incremental_unavailable` reports a service or work-budget failure. These failures do not advance progress. -The rollout gate is `BUZZ_ARTIFACT_SYNC_ENABLED`, off by default. Follow +The rollout gate is `BUZZ_ARTIFACT_INCREMENTAL_ENABLED`, off by default. Follow [artifact-sync-deployment.md](../artifact-sync-deployment.md) before enabling -it. Sync completion does not establish a live subscription to every channel. +it. Incremental completion does not establish a live subscription to every channel.