amadan
server+bot: removing a member tombstones their purged messages (#765) (#788)
* server+bot: removing a member tombstones their purged messages (#765) removeUser's cascade now inserts a tombstone per deleted message (same mechanism as deleteMessage's #581) before the DELETE — devices offline for the removal drop their local copies on the next forward catch-up, exactly when that propagation matters most. Legacy NULL-thread rows are skipped; threads torn down whole get none (their disappearance already clears every client). Kept surgical around the tamper-evidence journal call — and clear of the moderation stack's pending edits to the same file. Bot sidecar: ThreadCtx.History over-fetches (2x, server-capped at 200) and trims to the newest cputime unlimited filesize unlimited datasize unlimited stacksize 7MB coredumpsize 0kB addressspace unlimited memorylocked unlimited maxproc 10666 descriptors 1048576 real lines — tombstones, reactions and app events no longer silently shrink the model's context window; tombstone rows skip by field instead of a failed decrypt. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * server: journal the purge's tombstone inserts (tamper-evidence contract) Each purge tombstone is a real message row clients replay, so each gets a 'msg' journal row (set-based, same tx) — an unexplained row would read as tampering to a verifier. The join on the still-present doomed rows can't double-journal deleteMessage's older tombstones (their targets are already gone). Contract tests updated: db.go write-site count 7→8; the purge's pinned journal trace gains the two tombstone rows. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> ---------
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
4 files changed,
+104
−5
checkpoint_test.go+5 −2db.go+29 −0db_test.go+52 −0internal/bot/runtime.go+18 −3
diff --git a/checkpoint_test.go b/checkpoint_test.go| index 8e2cbd9..d3cc2b4 100644 |
| --- a/checkpoint_test.go |
| +++ b/checkpoint_test.go |
| @@ -110,7 +110,10 @@ func TestJournalCoversRemoveUserPurge(t *testing.T) { |
| for _, r := range journalRows(t, srv.db) { |
| kinds = append(kinds, r.Kind) |
| } |
| - want := []string{"msg", "msg", "delete", "delete"} |
| + // Two sends, their purge deletes, then the purge's tombstones (#765) — |
| + // real rows clients replay, so each carries its own 'msg' journal row |
| + // (an unexplained messages row would read as tampering to a verifier). |
| + want := []string{"msg", "msg", "delete", "delete", "msg", "msg"} |
| if len(kinds) != len(want) { |
| t.Fatalf("journal kinds %v, want %v", kinds, want) |
| } |
| @@ -438,7 +441,7 @@ func TestCheckpointColumnsMatchPayload(t *testing.T) { |
| // write site, update the expected counts here. |
| func TestMessageWritesAreJournaled(t *testing.T) { |
| want := map[string]int{ // file → number of messages-table write statements |
| - "db.go": 7, // insertMessage, editMessage, deleteMessage (delete + tombstone insert), removeUser ×2, deliverScheduledMessage |
| + "db.go": 8, // insertMessage, editMessage, deleteMessage (delete + tombstone insert), removeUser ×3 (purge tombstone insert #765 + delete ×2), deliverScheduledMessage |
| } |
| re := regexp.MustCompile(`INSERT INTO messages|UPDATE messages|DELETE FROM messages`) |
| files, err := filepath.Glob("*.go") |
diff --git a/db.go b/db.go| index 24682c3..465673f 100644 |
| --- a/db.go |
| +++ b/db.go |
| @@ -1600,6 +1600,35 @@ func (db *DB) removeUser(id string) error { |
| if err := journalMessageDeletesTx(tx, "author_hash = ?", id); err != nil { |
| return err |
| } |
| + // Tombstone each doomed message (#765, same mechanism as deleteMessage's |
| + // #581): members' devices that already hold these messages learn to drop |
| + // them from ordinary forward catch-up — a purge is exactly when that |
| + // propagation matters most. Skipped for legacy NULL-thread rows |
| + // (unreachable by thread reads); the fresh tombstones have a NULL |
| + // author_hash, so the cascade's DELETE below can't touch them. Threads |
| + // torn down whole (above) need none — the thread's disappearance already |
| + // clears every client's copy. |
| + purgeNow := time.Now().Unix() |
| + if _, err := tx.Exec( |
| + "INSERT INTO messages(created_at, blob, thread_id, tombstone_of) SELECT ?, '', thread_id, id FROM messages WHERE author_hash = ? AND thread_id IS NOT NULL", |
| + purgeNow, id, |
| + ); err != nil { |
| + return err |
| + } |
| + // Each tombstone is a real message row (clients replay it from catch-up), |
| + // so each gets a journal row like any other insert — an unexplained row |
| + // would read as tampering to a verifier. Set-based like the insert above: |
| + // the fresh tombstones are exactly those pointing at a still-present row |
| + // of the purged author (deleteMessage's older tombstones point at ids |
| + // already gone, so the join can't double-journal them). |
| + if _, err := tx.Exec( |
| + "INSERT INTO event_journal(ts, kind, msg_id, thread_id, blob_hash) "+ |
| + "SELECT ?, 'msg', t.id, COALESCE(t.thread_id, ''), ? FROM messages t "+ |
| + "JOIN messages d ON d.id = t.tombstone_of AND d.author_hash = ?", |
| + purgeNow, ciphertextHash(""), id, |
| + ); err != nil { |
| + return err |
| + } |
| for _, s := range []struct { |
| q string |
| arg []any |
diff --git a/db_test.go b/db_test.go| index 386b332..1ba5ed2 100644 |
| --- a/db_test.go |
| +++ b/db_test.go |
| @@ -90,6 +90,58 @@ func TestDeleteMessage(t *testing.T) { |
| } |
| } |
| +// Removing a member tombstones every message the purge deletes (#765): a |
| +// device that was offline for the removal must still learn to drop its local |
| +// copies — the exact propagation #581 built for single unsends, at the moment |
| +// it matters most. Threads torn down whole need (and keep) no tombstones. |
| +func TestRemoveUserTombstonesMessages(t *testing.T) { |
| + db := newTestDB(t) |
| + uidA, _ := db.createUser("Alice") |
| + uidB, _ := db.createUser("Bea") |
| + db.setUserPublicKey(uidA, "pub-a") |
| + db.setUserPublicKey(uidB, "pub-b") |
| + shared, err := db.createThread("group", "n", []ThreadMemberKey{ |
| + {UserID: uidA, Wrapped: "wa"}, {UserID: uidB, Wrapped: "wb"}, |
| + }) |
| + if err != nil { |
| + t.Fatalf("createThread: %v", err) |
| + } |
| + // author_hash stores the author's USER id (what hub.go passes) — the |
| + // purge matches on it, despite the column's historical name. |
| + m1, _ := db.insertMessage(shared, "bea-1", uidB) |
| + db.insertMessage(shared, "alice-1", uidA) |
| + m3, _ := db.insertMessage(shared, "bea-2", uidB) |
| + // A thread only Bea was in — torn down whole on removal, no tombstones. |
| + solo, err := db.createThread("self", "", []ThreadMemberKey{{UserID: uidB, Wrapped: "ws"}}) |
| + if err != nil { |
| + t.Fatalf("createThread solo: %v", err) |
| + } |
| + db.insertMessage(solo, "solo-note", uidB) |
| + |
| + if err := db.removeUser(uidB); err != nil { |
| + t.Fatalf("removeUser: %v", err) |
| + } |
| + rows, _ := db.messagesAfter(shared, 0, 100) |
| + var tombs []int64 |
| + for _, m := range rows { |
| + if m.TombstoneOf != 0 { |
| + tombs = append(tombs, m.TombstoneOf) |
| + if m.Blob != "" { |
| + t.Fatalf("tombstone %d carries a blob", m.ID) |
| + } |
| + } else if m.Blob != "alice-1" { |
| + t.Fatalf("Bea's ciphertext survived the purge: %+v", m) |
| + } |
| + } |
| + if len(tombs) != 2 || tombs[0] != m1.ID || tombs[1] != m3.ID { |
| + t.Fatalf("tombstones %v, want [%d %d]", tombs, m1.ID, m3.ID) |
| + } |
| + // The solo thread went entirely — including any would-be tombstones. |
| + if left, _ := db.messagesAfter(solo, 0, 100); len(left) != 0 { |
| + t.Fatalf("solo thread rows survived teardown: %+v", left) |
| + } |
| +} |
| + |
| // A delete must not read as fresh activity: the thread list's preview and |
| // recency ordering skip tombstone rows (#581) — deleting an old message |
| // neither blanks the preview nor bumps the thread above others. |
diff --git a/internal/bot/runtime.go b/internal/bot/runtime.go| index 153bdfb..f75440e 100644 |
| --- a/internal/bot/runtime.go |
| +++ b/internal/bot/runtime.go |
| @@ -126,16 +126,28 @@ type HistoryMsg struct { |
| func (tc *ThreadCtx) History(limit int) ([]HistoryMsg, error) { |
| var ml struct { |
| Messages []struct { |
| - ID int64 `json:"id"` |
| - Blob string `json:"blob"` |
| + ID int64 `json:"id"` |
| + Blob string `json:"blob"` |
| + TombstoneOf int64 `json:"tombstone_of"` |
| } `json:"messages"` |
| } |
| - path := fmt.Sprintf("/api/messages?thread=%s&limit=%d", url.QueryEscape(tc.ID), limit) |
| + // Over-fetch: delete tombstones (#581), reactions and app events ride the |
| + // window too, and each would otherwise shrink the model's context — a |
| + // thread with a burst of unsends could leave the brain near-blind (#765). |
| + // The server caps limit at 200; trim back to `limit` real lines below. |
| + fetch := limit * 2 |
| + if fetch > 200 { |
| + fetch = 200 |
| + } |
| + path := fmt.Sprintf("/api/messages?thread=%s&limit=%d", url.QueryEscape(tc.ID), fetch) |
| if err := tc.rt.get(path, &ml); err != nil { |
| return nil, err |
| } |
| var out []HistoryMsg |
| for _, m := range ml.Messages { |
| + if m.TombstoneOf != 0 { |
| + continue // an unsent message's marker, not a line anyone said |
| + } |
| var p struct { |
| Type string `json:"type"` |
| Name string `json:"name"` |
| @@ -150,6 +162,9 @@ func (tc *ThreadCtx) History(limit int) ([]HistoryMsg, error) { |
| Mine: p.Name == tc.rt.Session.Name, ActionCount: len(p.Actions), |
| }) |
| } |
| + if len(out) > limit { |
| + out = out[len(out)-limit:] // newest `limit` — oldest-first order kept |
| + } |
| return out, nil |
| } |