jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-13 12:51:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drop the one-shot member/bot roster reconcile scan

Browse Source 
Old rosters keep their stored label until the channel's next membership
change republishes them. Both labels grant the same permissions, and the
Huddle clients (#8544, #8545) recognize agents from managed-agent records,
so a stale label needs no operator step. This removes
BUZZ_RECONCILE_BOT_ROSTERS, its scan query and the env-flag rename.

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-10-27 11:03:35 -04:00
parent 9682cb157d
commit b0d05845e9
4 changed files with 20 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
crates/buzz-db/src/store/channel_members.rs
View File

@@ -1615,117 +1615,6 @@ pub async fn list_large_channel_rosters_needing_reconciliation(
.collect()
}
/// A channel whose live relay-authored kind:39002 roster shows an active
/// ordinary member with the other label (`member` instead of `bot`, or the
/// reverse) than [`member_role_label`] gives.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StaleRoleRoster {
/// Community that owns the channel.
pub community_id: CommunityId,
/// Canonical host for the owning community.
pub host: String,
/// Channel whose roster must be republished.
pub channel_id: Uuid,
}
/// Returns active channels whose live roster shows an ordinary member with
/// the wrong `member`/`bot` label.
///
/// Such a roster exists when it was published before this release (older
/// relays published the stored role), or when a relay stopped between
/// recording an agent owner and republishing that agent's rosters. Only two
/// kinds of channel can disagree, so only they are read: channels with an
/// active ordinary member that has a recorded agent owner, and channels whose
/// roster names someone `bot` (found through the `tags` GIN index). Only
/// active communities are scanned, as in
/// [`list_large_channel_rosters_needing_reconciliation`].
///
/// This is an internal cross-community maintenance read for the one-shot
/// `BUZZ_RECONCILE_BOT_ROSTERS` startup step. Callers must preserve the
/// returned community id when rewriting discovery state.
pub async fn list_channel_rosters_with_stale_ordinary_roles(
pool: &PgPool,
relay_pubkey: &[u8],
) -> Result<Vec<StaleRoleRoster>> {
let mut connection = crate::observability::acquire_writer(
pool,
crate::observability::WriterOperation::Maintenance,
)
.await?;
let rows = sqlx::query(concat!(
r#"
WITH candidates AS (
SELECT cm.community_id, cm.channel_id
FROM channel_members cm
JOIN users u
ON u.community_id = cm.community_id
AND u.pubkey = cm.pubkey
AND u.agent_owner_pubkey IS NOT NULL
WHERE cm.role IN ('member', 'bot') AND cm.removed_at IS NULL
UNION
SELECT roster.community_id, roster.channel_id
FROM events roster
WHERE roster.kind = 39002
AND roster.pubkey = $1
AND roster.deleted_at IS NULL
AND roster.channel_id IS NOT NULL
AND roster.tags @> '[["bot"]]'::jsonb
)
SELECT candidate.community_id, community.host, candidate.channel_id
FROM candidates candidate
JOIN channels ch
ON ch.community_id = candidate.community_id
AND ch.id = candidate.channel_id
AND ch.deleted_at IS NULL
JOIN communities community
ON community.id = candidate.community_id
AND community.archived_at IS NULL
AND community.deleted_at IS NULL
AND community.deletion_state = 'active'
JOIN LATERAL (
SELECT roster.tags
FROM events roster
WHERE roster.community_id = candidate.community_id
AND roster.channel_id = candidate.channel_id
AND roster.kind = 39002
AND roster.pubkey = $1
AND roster.deleted_at IS NULL
ORDER BY roster.created_at DESC, roster.id ASC
LIMIT 1
) live_roster ON true
WHERE EXISTS (
SELECT 1
FROM channel_members cm
JOIN jsonb_array_elements(live_roster.tags) tag
ON tag->>0 = 'p'
AND tag->>1 = encode(cm.pubkey, 'hex')
WHERE cm.community_id = candidate.community_id
AND cm.channel_id = candidate.channel_id
AND cm.removed_at IS NULL
AND cm.role IN ('member', 'bot')
AND tag->>3 IN ('member', 'bot')
AND tag->>3 <> "#,
member_role_label!(),
r#"
)
ORDER BY candidate.community_id, candidate.channel_id
"#
))
.bind(relay_pubkey)
.fetch_all(&mut *connection)
.await?;
rows.into_iter()
.map(|row| {
Ok(StaleRoleRoster {
community_id: CommunityId::from_uuid(row.try_get("community_id")?),
host: row.try_get("host")?,
channel_id: row.try_get("channel_id")?,
})
})
.collect()
}
/// The channels where `pubkey` is an active ordinary member (`member` or
/// `bot`). Recording an agent owner changes the label these rows show, so
/// their rosters are republished.
@@ -2388,19 +2277,6 @@ impl Db {
.await
}
/// Returns active-channel rosters that show an ordinary member with the
/// wrong `member`/`bot` label.
#[datastore_span(
name = "list_channel_rosters_with_stale_ordinary_roles",
system = "postgresql"
)]
pub async fn list_channel_rosters_with_stale_ordinary_roles(
&self,
relay_pubkey: &[u8],
) -> Result<Vec<StaleRoleRoster>> {
list_channel_rosters_with_stale_ordinary_roles(&self.pool, relay_pubkey).await
}
/// Returns the channels where `pubkey` is an active ordinary member.
#[datastore_span(name = "list_ordinary_member_channels", system = "postgresql")]
pub async fn list_ordinary_member_channels(
@@ -5407,11 +5283,9 @@ mod postgres_tests {
/// The roster check treats `member` and `bot` as one role, so an owner
/// recorded between roster capture and insert does not refuse the
/// publication; the background republish then replaces the stale label.
/// The one-shot scan reports a roster with a stale label until then.
///
/// Mutations: compare labels exactly in `validate_channel_roster_snapshot`
/// → the captured roster is refused → RED; compare stored roles in the
/// scan → the stale roster is not reported → RED.
/// Mutation: compare labels exactly in `validate_channel_roster_snapshot`
/// → the captured roster is refused → RED.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn owner_record_during_roster_publication_converges() {
@@ -5465,13 +5339,6 @@ mod postgres_tests {
.await
.expect("read live roster")
};
let stale = || async {
list_channel_rosters_with_stale_ordinary_roles(&pool, &relay_pubkey)
.await
.expect("list stale rosters")
.iter()
.any(|c| c.channel_id == channel)
};
let at = nostr::Timestamp::now().as_secs();
let mut snapshot = db
@@ -5497,7 +5364,6 @@ mod postgres_tests {
);
snapshot.release().await.expect("commit captured roster");
assert_eq!(agent_label(&live_tags().await).as_deref(), Some("member"));
assert!(stale().await, "the scan must report the stale label");
// The background republish after the owner record.
let mut snapshot = db
@@ -5511,7 +5377,6 @@ mod postgres_tests {
.expect("republish");
snapshot.release().await.expect("commit republish");
assert_eq!(agent_label(&live_tags().await).as_deref(), Some("bot"));
assert!(!stale().await);
}
/// The republish after an owner record lists the agent's active ordinary

33
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -3754,36 +3754,6 @@ pub async fn reconcile_large_channel_member_snapshots(
Ok(summary)
}
/// Republish rosters that show an ordinary member with the wrong
/// `member`/`bot` label: rosters published by relays from before the label
/// was computed, or left stale when a relay stopped between recording an
/// agent owner and republishing that agent's rosters.
///
/// A one-shot operator step, run at startup only when
/// `BUZZ_RECONCILE_BOT_ROSTERS` is set. Converged deployments find no
/// candidates, so it only reads.
pub async fn reconcile_ordinary_role_rosters(
state: &Arc<AppState>,
) -> anyhow::Result<ReconcileSummary> {
let relay_pubkey = state.relay_keypair.public_key();
let candidates = state
.db
.list_channel_rosters_with_stale_ordinary_roles(&relay_pubkey.to_bytes())
.await?;
let summary = republish_roster_candidates(
state,
candidates
.into_iter()
.map(|candidate| (candidate.community_id, candidate.host, candidate.channel_id))
.collect(),
"ordinary role roster reconciliation failed",
)
.await;
metrics::counter!("buzz_channel_role_roster_reconciliations_total")
.increment(summary.repaired as u64);
Ok(summary)
}
/// Republish each `(community, host, channel)` roster under its membership
/// lock, counting failures instead of stopping. Each roster first takes a
/// relay-wide republish permit, so concurrent callers share a small bound on
@@ -3847,8 +3817,7 @@ async fn republish_roster_candidates(
/// wait: first AUTH must finish inside its own deadline. A failure is logged
/// and counted and does not stop the other channels. A roster left stale (for
/// example by a relay that stopped first) shows only the wrong label, which
/// has the same permissions; the next membership change or a
/// `BUZZ_RECONCILE_BOT_ROSTERS` startup repairs it.
/// has the same permissions; the next membership change repairs it.
pub fn spawn_agent_roster_republish(tenant: &TenantContext, state: Arc<AppState>, agent: Vec<u8>) {
let (community_id, host) = (tenant.community(), tenant.host().to_owned());
tokio::spawn(async move {

57
crates/buzz-relay/src/main.rs
View File

@@ -30,9 +30,7 @@ use buzz_relay::telemetry;
use buzz_workflow::WorkflowEngine;
use tokio_util::sync::CancellationToken;
/// Whether an opt-in boolean environment flag is on: `true`, `1`, `yes` or
/// `on`, in any case. Unset, empty and every other value mean off.
fn env_flag_enabled(value: Option<&str>) -> bool {
fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool {
value.map(str::trim).is_some_and(|value| {
matches!(
value.to_ascii_lowercase().as_str(),
@@ -547,7 +545,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
info!("Postgres connected");
}
let auto_migrate = env_flag_enabled(std::env::var("BUZZ_AUTO_MIGRATE").ok().as_deref());
let auto_migrate =
buzz_auto_migrate_enabled(std::env::var("BUZZ_AUTO_MIGRATE").ok().as_deref());
if auto_migrate {
let step = StepTimer::start(StartupStep::DbMigrate);
db.migrate().await.map_err(|e| {
@@ -1150,30 +1149,6 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
}
step.finish();
// One-shot operator step after upgrading to computed `member`/`bot`
// labels: republish rosters that older relays published with the stored
// label. Off by default so no replica scans at every startup; both labels
// have the same permissions, so a stale one is only cosmetic until then.
if env_flag_enabled(std::env::var("BUZZ_RECONCILE_BOT_ROSTERS").ok().as_deref()) {
let role_roster_state = Arc::clone(&state);
tokio::spawn(async move {
match buzz_relay::handlers::side_effects::reconcile_ordinary_role_rosters(
&role_roster_state,
)
.await
{
Ok(summary) => info!(
count = summary.repaired,
failed = summary.failed,
"member/bot channel rosters reconciled"
),
Err(error) => {
tracing::warn!(%error, "member/bot channel roster reconciliation failed")
}
}
});
}
// Emit kind:39000/39002 discovery events for channels that exist in the DB
// but don't have corresponding events (e.g. seeded via direct SQL inserts).
// Only runs when BUZZ_RECONCILE_CHANNELS=true (dev/CI environments).
@@ -3138,8 +3113,8 @@ mod tests {
use uuid::Uuid;
use super::{
apply_activity_refresh, apply_stock_refresh, connect_audit_pool, counted_partition_audit,
dropped_in_memory_keys, emit_cached_fleet_usage_metrics, env_flag_enabled,
apply_activity_refresh, apply_stock_refresh, buzz_auto_migrate_enabled, connect_audit_pool,
counted_partition_audit, dropped_in_memory_keys, emit_cached_fleet_usage_metrics,
idle_timeout_secs, jwks_next_retry_after_failed_refresh, nip_fi_jwks_refresh_loop,
record_legacy_usage_collection, refresh_due_fleet_usage,
refresh_legacy_active_gauge_recency, relay_keypair_from_config,
@@ -3508,18 +3483,18 @@ mod tests {
}
#[test]
fn env_flags_are_opt_in() {
assert!(!env_flag_enabled(None));
assert!(!env_flag_enabled(Some("")));
assert!(!env_flag_enabled(Some("false")));
assert!(!env_flag_enabled(Some("0")));
assert!(!env_flag_enabled(Some("no")));
fn buzz_auto_migrate_is_opt_in() {
assert!(!buzz_auto_migrate_enabled(None));
assert!(!buzz_auto_migrate_enabled(Some("")));
assert!(!buzz_auto_migrate_enabled(Some("false")));
assert!(!buzz_auto_migrate_enabled(Some("0")));
assert!(!buzz_auto_migrate_enabled(Some("no")));
assert!(env_flag_enabled(Some("true")));
assert!(env_flag_enabled(Some("TRUE")));
assert!(env_flag_enabled(Some(" 1 ")));
assert!(env_flag_enabled(Some("yes")));
assert!(env_flag_enabled(Some("on")));
assert!(buzz_auto_migrate_enabled(Some("true")));
assert!(buzz_auto_migrate_enabled(Some("TRUE")));
assert!(buzz_auto_migrate_enabled(Some(" 1 ")));
assert!(buzz_auto_migrate_enabled(Some("yes")));
assert!(buzz_auto_migrate_enabled(Some("on")));
}
#[test]

2
deploy/charts/buzz/README.md
View File

@@ -616,7 +616,7 @@ default so long-lived WebSocket connections have time to drain.
Schema migrations are embedded in the relay binary via `sqlx::migrate!` and run at startup, gated by `BUZZ_AUTO_MIGRATE` (default `true`). Multiple replicas race-safely behind a Postgres advisory lock. `helm upgrade` is the entire upgrade procedure.
Migration 0032 is a hard compatibility boundary for relay versions that publish repaired channel rosters. Older relays verify its roster-fence trigger before opening listeners and refuse to start if the trigger is missing or inert. Newer relays run the same check in every roster writer and do not require the trigger. Migration 0074 drops the trigger, so every relay must run a version without the startup check (#8611 or later) before 0074 is applied; an older relay then refuses to start. Relays from the 0074 release on show a computed `member`/`bot` label in rosters that the trigger would refuse, so they refuse to start while the trigger is installed. Apply 0074 before rolling them out. After that rollout, start one relay once with `BUZZ_RECONCILE_BOT_ROSTERS=1` (`true`, `yes` and `on` also work; any other value leaves it off) to republish rosters that older relays published with the stored label. Confirm that it logs `member/bot channel rosters reconciled`; a warning instead means the repair did not run, so start it again. Apply migrations before rolling the relay; for large installations, prefer a controlled `buzz-admin migrate` job with PostgreSQL lock monitoring before the code rollout.
Migration 0032 is a hard compatibility boundary for relay versions that publish repaired channel rosters. Older relays verify its roster-fence trigger before opening listeners and refuse to start if the trigger is missing or inert. Newer relays run the same check in every roster writer and do not require the trigger. Migration 0074 drops the trigger, so every relay must run a version without the startup check (#8611 or later) before 0074 is applied; an older relay then refuses to start. Relays from the 0074 release on show a computed `member`/`bot` label in rosters that the trigger would refuse, so they refuse to start while the trigger is installed. Apply 0074 before rolling them out. A roster that an older relay published keeps its stored label until the channel's next membership change republishes it. Both labels have the same permissions. Apply migrations before rolling the relay; for large installations, prefer a controlled `buzz-admin migrate` job with PostgreSQL lock monitoring before the code rollout.
If you prefer decoupling migrations from serving, set `migrate.autoMigrate=false`. **In that mode the chart does not run migrations for you** — you own running `buzz-admin migrate` (separate Pod / one-shot Job) against the database before every `helm install` / `helm upgrade`. Readiness probes only verify DB connectivity, not schema freshness, so a pod will appear healthy against an unmigrated schema and fail under load. A pre-upgrade Helm Job for this is on the chart roadmap; the values knob `migrate.preUpgradeJob.enabled` is reserved.