Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -262,56 +262,42 @@ async fn sync_shrinks_large_pages_and_current_batches_fail_explicitly() {

#[tokio::test]
#[ignore = "requires Postgres"]
async fn sync_gate_controls_http_reads_and_capability() {
async fn sync_http_reads_and_capability_are_available() {
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_incremental_enabled =
enabled;
let response = crate::router::build_router(f.state.clone())
.oneshot(
axum::http::Request::builder()
.uri("/info")
.header("host", &f.host)
.body(axum::body::Body::empty())
.unwrap(),
)
.await
.unwrap();
assert!(response.status().is_success());
let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024)
.await
.unwrap();
let info: Value = serde_json::from_slice(&bytes).unwrap();
if enabled {
assert_eq!(info["artifacts"]["sync"], websocket_sync);
assert_eq!(
info["artifacts"]["incremental"],
json!({"version":1,"overlapSeconds":120,
"pageRows":100,"currentRows":100,"responseBytes":8388608,
"permanentChannelsOnly":true,"checkpointLifetimeSeconds":2592000})
);
} else {
websocket_sync = info["artifacts"]["sync"].clone();
assert!(info["artifacts"].get("incremental").is_none());
}
let (status, _) = f
.request(
"/query",
json!([{"artifact":"incremental","query":query()}]),
)
.await;
assert_eq!(
status,
if enabled {
StatusCode::OK
} else {
StatusCode::SERVICE_UNAVAILABLE
}
);
}
let f = fixture().await;
let response = crate::router::build_router(f.state.clone())
.oneshot(
axum::http::Request::builder()
.uri("/info")
.header("host", &f.host)
.body(axum::body::Body::empty())
.unwrap(),
)
.await
.unwrap();
assert!(response.status().is_success());
let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024)
.await
.unwrap();
let info: Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(
info["artifacts"]["sync"],
json!({"version":1,"max_events":4096})
);
assert_eq!(
info["artifacts"]["incremental"],
json!({"version":1,"overlapSeconds":120,
"pageRows":100,"currentRows":100,"responseBytes":8388608,
"permanentChannelsOnly":true,"checkpointLifetimeSeconds":2592000})
);
let (status, _) = f
.request(
"/query",
json!([{"artifact":"incremental","query":query()}]),
)
.await;
assert_eq!(status, StatusCode::OK);
}

#[tokio::test]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,7 @@ mod policy_postgres_tests;
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_incremental_enabled = true;
f.state = Arc::new(state);
f
Fixture::new().await
}
static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,6 @@ 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_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());
Expand Down
3 changes: 0 additions & 3 deletions crates/buzz-relay/src/api/artifact_sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -125,9 +125,6 @@ pub(super) async fn query(
reader: &nostr::PublicKey,
filter: &Value,
) -> Response {
if !state.config.artifact_incremental_enabled {
return Err(unavailable());
}
let _permit = WORK.try_acquire().map_err(|_| unavailable())?;
tokio::time::timeout(
std::time::Duration::from_secs(8),
Expand Down
6 changes: 0 additions & 6 deletions crates/buzz-relay/src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -116,8 +116,6 @@ pub const MAX_DRAIN_JITTER_MS: u64 = 20_000;
/// Relay runtime configuration, loaded from environment variables.
#[derive(Debug, Clone)]
pub struct Config {
/// Enable incremental artifact reads after every serving writer is upgraded.
pub artifact_incremental_enabled: bool,
/// Address the relay HTTP/WebSocket server binds to.
pub bind_addr: SocketAddr,
/// Postgres database connection URL.
Expand Down Expand Up @@ -799,9 +797,6 @@ impl Config {
let require_relay_membership = std::env::var("BUZZ_REQUIRE_RELAY_MEMBERSHIP")
.map(|v| v == "true" || v == "1")
.unwrap_or(false);
let artifact_incremental_enabled = std::env::var("BUZZ_ARTIFACT_INCREMENTAL_ENABLED")
.map(|v| v == "true" || v == "1")
.unwrap_or(false);

// Defaults true → single-pod (N=1) keeps today's huddle behavior. A
// horizontally-scaled deployment sets this false; see the field doc.
Expand Down Expand Up @@ -1457,7 +1452,6 @@ impl Config {
partition_manager_advance_enabled,
pubkey_allowlist_enabled,
require_relay_membership,
artifact_incremental_enabled,
huddle_audio_available,
mesh,
mesh_demo_echo,
Expand Down
4 changes: 1 addition & 3 deletions crates/buzz-relay/src/nip11.rs
Original file line number Diff line number Diff line change
Expand Up @@ -395,9 +395,7 @@ 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_incremental_enabled {
info.artifacts["incremental"] = crate::api::artifact_sync::capability();
}
info.artifacts["incremental"] = crate::api::artifact_sync::capability();
if let Ok(tenant) = crate::tenant::bind_community(state, raw_host).await {
if let Some(profile) = crate::verified_profiles::discovery(state, &tenant).await {
if let Some(fi) = info.federated_identity.as_mut() {
Expand Down
25 changes: 7 additions & 18 deletions docs/artifact-sync-deployment.md
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
# Incremental artifact reads deployment

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.
timestamps. The schema, indexes, and every serving writer must support
incremental reads.

## Schema and indexes

Expand Down Expand Up @@ -49,22 +49,11 @@ After all index builds succeed, run `buzz-admin migrate` with the target
then verifies and records `0078`. Direct SQL application does not record the
migration ledger.

## Enable and rollback
## Serving fleet

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_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
instance. Before enabling it again, upgrade every writer and advance
`communities.artifact_policy_modified_at` for each affected active community
in a reviewed maintenance transaction under community deletion admission.
This forces clients to discover current records and revalidate saved IDs.
No local absence scan is permitted.
All instances must use the same configured relay signing key, because that key
authenticates the progress tokens. NIP-11 advertises `artifacts.incremental`.
Every serving writer must maintain the modification timestamps.

The 120-second stamp-to-visibility bound remains an operating assumption.
Runtime measurements are tracked separately. This PR does not authorize
a deployment or a production maintenance write.
Runtime measurements are tracked separately.
4 changes: 1 addition & 3 deletions docs/nips/NIP-AR.md
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,4 @@ 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_INCREMENTAL_ENABLED`, off by default. Follow
[artifact-sync-deployment.md](../artifact-sync-deployment.md) before enabling
it. Incremental completion does not establish a live subscription to every channel.
Incremental completion does not establish a live subscription to every channel.
Loading