fix(desktop): defer capability changes until real time advances

Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-09-04 12:24:55 -04:00
parent 0fdf436105
commit 09282c7283
4 changed files with 141 additions and 10 deletions

View File

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

View File

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

View File

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

View File

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