jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 20:41:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): name incremental artifact reads

Browse Source 
Signed-off-by: Daniel Rouleau <drouleau@squareup.com>
This commit is contained in:
Daniel Rouleau
2026-10-22 15:06:19 -07:00
parent babcdb7664
commit 97cacecb5c
13 changed files with 105 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/buzz-db/src/store/artifact_query.rs
View File

@@ -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<sqlx::Postgres>,
community: CommunityId,

2
crates/buzz-db/src/store/artifact_sync.rs
View File

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

7
crates/buzz-relay/src/api/artifact.rs
View File

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

54
crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs
View File

@@ -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]}]),
)

6
crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs
View File

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

6
crates/buzz-relay/src/api/artifact_postgres_tests/sync_policy_postgres_tests.rs
View File

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

32
crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs
View File

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

6
crates/buzz-relay/src/api/artifact_postgres_tests/sync_scale_postgres_tests.rs
View File

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

27
crates/buzz-relay/src/api/artifact_sync.rs
View File

@@ -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<Value>) {
(status, Json(json!({"error":code,"code":code})))
}
fn invalid() -> (StatusCode, Json<Value>) {
error(StatusCode::BAD_REQUEST, "sync_invalid")
error(StatusCode::BAD_REQUEST, "incremental_invalid")
}
fn unavailable() -> (StatusCode, Json<Value>) {
error(StatusCode::SERVICE_UNAVAILABLE, "sync_unavailable")
error(StatusCode::SERVICE_UNAVAILABLE, "incremental_unavailable")
}
fn limit() -> (StatusCode, Json<Value>) {
error(StatusCode::PAYLOAD_TOO_LARGE, "sync_limit")
error(StatusCode::PAYLOAD_TOO_LARGE, "incremental_limit")
}
fn reset(reason: &str) -> (StatusCode, Json<Value>) {
(
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<Value>) {
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<Hmac<Sha256>, (StatusCode, Json<Value>)> {
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::<Sha256>::new_from_slice(&digest.finalize()).map_err(|_| unavailable())
}
@@ -120,7 +121,7 @@ fn decode(state: &AppState, value: &str) -> Result<Token, (StatusCode, Json<Valu
serde_json::from_slice(&data).map_err(|_| invalid())
}
/// Execute one sync operation without a manifest or server-side cursor state.
/// Execute one incremental read without a manifest or server-side cursor state.
#[tracing::instrument(skip_all, fields(community_id = %tenant.community(), reader = %reader))]
pub(super) async fn query(
state: &Arc<AppState>,
@@ -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()
{

6
crates/buzz-relay/src/config.rs
View File

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

4
crates/buzz-relay/src/nip11.rs
View File

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

10
docs/artifact-sync-deployment.md
View File

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

28
docs/nips/NIP-AR.md
View File

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