mirror of
https://github.com/block/buzz.git
synced 2026-10-29 05:21:31 +00:00
fix(desktop): defer capability changes until real time advances
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
@@ -75,6 +75,11 @@ impl DesktopCapabilities {
|
||||
|
||||
/// Encrypt/sign once, then persist these exact bytes for retries.
|
||||
pub fn sign(&self, keys: &Keys) -> Result<Event, String> {
|
||||
self.sign_at(keys, nostr::Timestamp::now())
|
||||
}
|
||||
|
||||
/// Sign at an observed wall-clock second, never a synthesized logical time.
|
||||
pub fn sign_at(&self, keys: &Keys, observed: nostr::Timestamp) -> Result<Event, String> {
|
||||
self.validate()?;
|
||||
let content = nip44::encrypt(
|
||||
keys.secret_key(),
|
||||
@@ -85,6 +90,7 @@ impl DesktopCapabilities {
|
||||
.map_err(|e| e.to_string())?;
|
||||
let event = EventBuilder::new(Kind::Custom(KIND_DESKTOP_CAPABILITIES as u16), content)
|
||||
.tag(Tag::identifier(&self.id))
|
||||
.custom_created_at(observed)
|
||||
.sign_with_keys(keys)
|
||||
.map_err(|e| e.to_string())?;
|
||||
validate_envelope(&event)?;
|
||||
|
||||
@@ -50,13 +50,16 @@ pub async fn prepare_desktop_capabilities(
|
||||
let facts =
|
||||
project(super::agent_discovery::discover_acp_providers(app.clone(), Some(false)).await?)?;
|
||||
let scope = scope(&app, &state, &owner, &community)?;
|
||||
Ok(json!({ "event": prepare_report(&mut open_retention_db(&scope.db_path)?, &scope, facts)? }))
|
||||
Ok(
|
||||
json!({ "event": prepare_report(&mut open_retention_db(&scope.db_path)?, &scope, facts, nostr::Timestamp::now)? }),
|
||||
)
|
||||
}
|
||||
|
||||
fn prepare_report(
|
||||
conn: &mut Connection,
|
||||
scope: &RetentionScope,
|
||||
facts: Vec<RuntimeFact>,
|
||||
clock: impl FnOnce() -> nostr::Timestamp,
|
||||
) -> Result<Event, String> {
|
||||
let saved = prepare(conn, scope)?;
|
||||
let profile: Event =
|
||||
@@ -90,8 +93,16 @@ fn prepare_report(
|
||||
.unwrap_or(false);
|
||||
let event = match previous {
|
||||
Some(event) if unchanged => event,
|
||||
_ => {
|
||||
let event = report.sign(&scope.owner_keys)?;
|
||||
previous => {
|
||||
let now = clock();
|
||||
// Keep the prior retry record until real time advances. Signing tied
|
||||
// ciphertext can lose NIP-33's lower-ID tie; never cache that loss or
|
||||
// future-date a replacement. The existing pulse/reconnect/Refresh
|
||||
// retries discovery, not a captured projection, without waiting here.
|
||||
if previous.as_ref().is_some_and(|e| now <= e.created_at) {
|
||||
return Err("Desktop capability facts deferred until the clock advances".into());
|
||||
}
|
||||
let event = report.sign_at(&scope.owner_keys, now)?;
|
||||
tx.execute(
|
||||
"INSERT OR REPLACE INTO desktop_capabilities VALUES (1, ?1)",
|
||||
[event.as_json()],
|
||||
@@ -123,7 +134,7 @@ pub fn read_desktop_capabilities(
|
||||
mod tests {
|
||||
use super::*;
|
||||
#[test]
|
||||
fn unchanged_facts_reopen_exact_bytes_changed_facts_replace_atomically() {
|
||||
fn changed_facts_defer_until_real_clock_advances_then_win_signed_order() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let scope = RetentionScope {
|
||||
db_path: dir.path().join("report.db"),
|
||||
@@ -134,26 +145,80 @@ mod tests {
|
||||
&mut open_retention_db(&scope.db_path).unwrap(),
|
||||
&scope,
|
||||
vec![],
|
||||
|| nostr::Timestamp::from(1000),
|
||||
)
|
||||
.unwrap();
|
||||
let mut reopened = open_retention_db(&scope.db_path).unwrap();
|
||||
assert_eq!(
|
||||
prepare_report(&mut reopened, &scope, vec![]).unwrap(),
|
||||
prepare_report(&mut reopened, &scope, vec![], || panic!(
|
||||
"unchanged must not sign"
|
||||
))
|
||||
.unwrap(),
|
||||
first
|
||||
);
|
||||
assert_eq!(reopened.total_changes(), 0);
|
||||
let facts = vec![RuntimeFact {
|
||||
let mut facts = vec![RuntimeFact {
|
||||
id: "goose".into(),
|
||||
availability: "available".into(),
|
||||
requires_external_cli: true,
|
||||
max_parallelism: None,
|
||||
}];
|
||||
let changed = prepare_report(&mut reopened, &scope, facts).unwrap();
|
||||
assert_ne!(changed.id, first.id);
|
||||
for now in [1000, 990, 999, 1000] {
|
||||
let error = prepare_report(&mut reopened, &scope, facts.clone(), || {
|
||||
nostr::Timestamp::from(now)
|
||||
})
|
||||
.unwrap_err();
|
||||
assert!(error.contains("clock advances"));
|
||||
assert_eq!(reopened.total_changes(), 0, "deferral must not persist");
|
||||
// Returning to old facts cancels the proposed change, even after a
|
||||
// restart/rollback: no deferred payload or timestamp renewal survives.
|
||||
assert_eq!(
|
||||
prepare_report(&mut reopened, &scope, vec![], || panic!("exact retry")).unwrap(),
|
||||
first
|
||||
);
|
||||
reopened = open_retention_db(&scope.db_path).unwrap();
|
||||
}
|
||||
// The retry observes today's facts, not the projection first deferred.
|
||||
facts[0].availability = "cli_missing".into();
|
||||
let changed = prepare_report(&mut reopened, &scope, facts.clone(), || {
|
||||
nostr::Timestamp::from(1001)
|
||||
})
|
||||
.unwrap();
|
||||
first.verify().unwrap();
|
||||
changed.verify().unwrap();
|
||||
assert_eq!(changed.created_at.as_secs(), 1001, "no future timestamp");
|
||||
assert!(changed.created_at > first.created_at);
|
||||
assert_eq!(changed.tags, first.tags);
|
||||
for events in [
|
||||
vec![first.clone(), changed.clone()],
|
||||
vec![changed.clone(), first],
|
||||
] {
|
||||
let rows =
|
||||
DesktopCapabilities::read_latest(events, &scope.owner_keys, &scope.relay_url)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 1);
|
||||
assert_eq!(rows[0].0.runtimes, facts);
|
||||
assert_eq!(rows[0].1, 1001);
|
||||
}
|
||||
let mut reopened = open_retention_db(&scope.db_path).unwrap();
|
||||
let mut invalid = facts.clone();
|
||||
invalid[0].max_parallelism = Some(0);
|
||||
assert!(prepare_report(&mut reopened, &scope, invalid, || {
|
||||
nostr::Timestamp::from(1002)
|
||||
})
|
||||
.is_err());
|
||||
assert_eq!(
|
||||
prepare_report(&mut reopened, &scope, facts, || panic!("exact retry")).unwrap(),
|
||||
changed
|
||||
);
|
||||
assert_eq!(
|
||||
reopened.total_changes(),
|
||||
0,
|
||||
"failed signing must not persist"
|
||||
);
|
||||
reopened
|
||||
.execute("UPDATE desktop_capabilities SET raw = 'corrupt'", [])
|
||||
.unwrap();
|
||||
assert!(prepare_report(&mut reopened, &scope, vec![]).is_err());
|
||||
assert!(prepare_report(&mut reopened, &scope, vec![], nostr::Timestamp::now).is_err());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -80,6 +80,34 @@ test("unchanged accepted report does not republish; failed publish retries exact
|
||||
await assert.rejects(f.refresh(), /unavailable/);
|
||||
});
|
||||
|
||||
test("deferred preparation settles with prior relay facts, never publishes, and honors cancellation", async () => {
|
||||
const f = fixture();
|
||||
const ipc = f.ipc;
|
||||
let active = true;
|
||||
let cancel = false;
|
||||
f.ipc = async (command, args) => {
|
||||
if (command === "prepare_desktop_capabilities") {
|
||||
if (cancel) active = false;
|
||||
throw Error("Desktop capability facts deferred until the clock advances");
|
||||
}
|
||||
return ipc(command, args);
|
||||
};
|
||||
const deferred = await f.refresh(() => active);
|
||||
assert.deepEqual(deferred.rows, [row]);
|
||||
assert.match(deferred.warning, /Will retry/);
|
||||
assert.ok(!f.calls.includes(event));
|
||||
cancel = true;
|
||||
await assert.rejects(
|
||||
f.refresh(() => active),
|
||||
/scope changed/,
|
||||
);
|
||||
assert.ok(!f.calls.includes(event));
|
||||
// A later, active attempt prepares afresh; no held promise or queued event.
|
||||
f.ipc = ipc;
|
||||
assert.equal((await f.refresh()).warning, "");
|
||||
assert.equal(f.calls.filter((c) => c === event).length, 1);
|
||||
});
|
||||
|
||||
test("all async boundaries fence cancellation, account/community switches and late ACK", async () => {
|
||||
for (const boundary of ["prepare", "read", "fetch", "transport", "ack"]) {
|
||||
const f = fixture(boundary);
|
||||
|
||||
@@ -202,6 +202,9 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa
|
||||
const originalReconnect = relayClient.subscribeToReconnects;
|
||||
let reconnect;
|
||||
let pulses = 0;
|
||||
let reports = 0;
|
||||
let publishedReports = 0;
|
||||
let deferReport = true;
|
||||
relayClient.subscribeToReconnects = (callback) => {
|
||||
reconnect = callback;
|
||||
return () => {
|
||||
@@ -210,6 +213,17 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa
|
||||
};
|
||||
window.__TAURI_INTERNALS__ = {
|
||||
invoke: async (command, args) => {
|
||||
if (command === "prepare_desktop_capabilities") {
|
||||
reports++;
|
||||
if (deferReport) throw Error("clock has not advanced");
|
||||
return { event: { ...first, kind: 30182 } };
|
||||
}
|
||||
if (command === "read_desktop_capabilities")
|
||||
return args.events.map(() => ({
|
||||
id: "desktop-a",
|
||||
reported: 100,
|
||||
runtimes: [],
|
||||
}));
|
||||
if (command === "prepare_desktop_observation")
|
||||
return { event: { ...first, kind: 30181 } };
|
||||
if (command === "read_desktop_observations")
|
||||
@@ -237,7 +251,7 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa
|
||||
};
|
||||
relayClient.fetchEvents = async (filter) => {
|
||||
if (fail) throw Error("unavailable");
|
||||
if (filter.kinds[0] === 30181) return [];
|
||||
if ([30181, 30182].includes(filter.kinds[0])) return [];
|
||||
if (filter["#d"]) return [current];
|
||||
const rows = [current];
|
||||
if (hold) {
|
||||
@@ -249,6 +263,10 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa
|
||||
return rows;
|
||||
};
|
||||
relayClient.publishEvent = async (event) => {
|
||||
if (event.kind === 30182) {
|
||||
publishedReports++;
|
||||
return;
|
||||
}
|
||||
assert.equal(event.kind, 30181, "no profile heartbeat rewrite");
|
||||
pulses++;
|
||||
};
|
||||
@@ -285,14 +303,28 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa
|
||||
await settle();
|
||||
assert.match(text(), /owner-a-wss:\/\/a.example/);
|
||||
assert.match(text(), /Last heard: Recent/);
|
||||
assert.match(text(), /could not synchronize capability facts/);
|
||||
const beforeReconnect = pulses;
|
||||
const reportsBeforeReconnect = reports;
|
||||
await React.act(async () => reconnect());
|
||||
await settle();
|
||||
assert.ok(pulses > beforeReconnect, "reconnect reports a fresh pulse");
|
||||
assert.ok(
|
||||
reports > reportsBeforeReconnect,
|
||||
"reconnect retries deferred facts",
|
||||
);
|
||||
const beforeTimer = pulses;
|
||||
const reportsBeforeTimer = reports;
|
||||
await React.act(async () => t.mock.timers.tick(60_000));
|
||||
await settle();
|
||||
assert.ok(pulses > beforeTimer, "bounded periodic publisher runs");
|
||||
assert.ok(reports > reportsBeforeTimer, "periodic retry survives deferral");
|
||||
assert.equal(publishedReports, 0, "deferred facts are not published");
|
||||
deferReport = false;
|
||||
await React.act(async () => t.mock.timers.tick(60_000));
|
||||
await settle();
|
||||
assert.equal(publishedReports, 1, "later preparation is published");
|
||||
assert.doesNotMatch(text(), /could not synchronize capability facts/);
|
||||
hold = true;
|
||||
await React.act(async () => {
|
||||
void client.refetchQueries({ queryKey: ["desktop-profiles"] });
|
||||
|
||||
Reference in New Issue
Block a user