amadan
server: record a delete tombstone so offline clients replay unsends (#581) (#763)
* server: record a delete tombstone so offline clients replay unsends (#581) Deleting a message keeps the hard delete (ciphertext really goes) and, in the same transaction, inserts a fresh empty row with tombstone_of = the deleted id. The fresh autoincrement id lands above every client's forward watermark, so the ordinary after: catch-up delivers the delete to devices that missed the live broadcast — no new sync machinery. - additive migration: messages.tombstone_of INTEGER - /api/messages returns tombstone_of (omitempty; old clients drop the empty-blob row, so behaviour is unchanged until clients fold it) - thread-list preview + recency ordering skip tombstones (a delete is bookkeeping, not activity) - README: disclosed-posture line (a durable dated record that a deletion happened now exists) Client folds are follow-ups: web #583 (with the chat.json event vector), native sibling for the iOS/Mac stores. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * server: harden delete tombstone against raced duplicates + legacy NULL-thread rows Review findings: gate the tombstone INSERT on the DELETE's RowsAffected (restores the old 'true = this call removed the row' invariant; a raced duplicate mints no second tombstone and triggers no second broadcast), skip tombstones for legacy NULL-thread_id rows (unreachable by any thread read), and document why reaper's has_content deliberately counts tombstones. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> ---------
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
5 files changed,
+164
−17
README.md+6 −0db.go+61 −12db_test.go+67 −5handlers_test.go+26 −0reaper.go+4 −0
diff --git a/README.md b/README.md| index 17446fa..68ad4c8 100644 |
| --- a/README.md |
| +++ b/README.md |
| @@ -41,6 +41,12 @@ see Trade-offs.) |
| copy carried inside each message), and the server necessarily knows who shares |
| a thread with whom in order to route messages. So a server compromise reveals |
| the social graph and names — just not what anyone said. |
| +- **Unsending leaves a dated marker.** Deleting a message really deletes the |
| + ciphertext, but the server keeps a small tombstone row — which thread, which |
| + message id, when — so members' devices that were offline at the time still |
| + learn to drop their local copy. The tombstone holds no content and no key |
| + material, but it is a durable record *that* a deletion happened (the server |
| + transiently knew that anyway — it performed the delete). |
| - **Key distribution trusts the server.** Public keys are handed out by the |
| server's roster and aren't yet verified out of band (no safety-number check). |
| A *passive* compromise — a stolen database — can't use that (the content |
diff --git a/db.go b/db.go| index 5eef5ac..ecfc79d 100644 |
| --- a/db.go |
| +++ b/db.go |
| @@ -692,6 +692,14 @@ CREATE TABLE IF NOT EXISTS member_bots ( |
| // so a redelivered or out-of-order webhook (Stripe is at-least-once and |
| // can reorder) is ignored rather than resurrecting a revoked membership. |
| "ALTER TABLE user_tiers ADD COLUMN ext_event_ts INTEGER NOT NULL DEFAULT 0", |
| + // Delete tombstones (#581): unsending hard-deletes the row (the |
| + // ciphertext really goes) and inserts a fresh empty row whose |
| + // tombstone_of names the deleted id. The fresh autoincrement id lands |
| + // above every client's forward-sync watermark, so offline clients |
| + // replay the delete from ordinary catch-up — the live broadcast stays |
| + // the instant path. Stores only (thread, target id, deleted-at): |
| + // metadata the server transiently knew when it performed the delete. |
| + "ALTER TABLE messages ADD COLUMN tombstone_of INTEGER", |
| } { |
| if _, err := db.sql.Exec(c); err != nil && !strings.Contains(err.Error(), "duplicate column") { |
| return err |
| @@ -2519,8 +2527,10 @@ func (db *DB) wrappedKeyFor(threadID, userID string) (string, bool, error) { |
| func (db *DB) threadsForUser(userID string) ([]Thread, error) { |
| rows, err := db.sql.Query(` |
| SELECT t.id, t.kind, COALESCE(t.name_blob, ''), t.created_at, COALESCE(k.wrapped, ''), |
| - COALESCE((SELECT blob FROM messages WHERE thread_id = t.id ORDER BY id DESC LIMIT 1), ''), |
| - COALESCE((SELECT created_at FROM messages WHERE thread_id = t.id ORDER BY id DESC LIMIT 1), 0), |
| + -- tombstone rows (#581) are sync bookkeeping, not activity: they must |
| + -- not become the preview blob or bump the thread's recency below. |
| + COALESCE((SELECT blob FROM messages WHERE thread_id = t.id AND tombstone_of IS NULL ORDER BY id DESC LIMIT 1), ''), |
| + COALESCE((SELECT created_at FROM messages WHERE thread_id = t.id AND tombstone_of IS NULL ORDER BY id DESC LIMIT 1), 0), |
| COALESCE(t.avatar_blob, ''), |
| COALESCE(t.visibility, 'invite'), COALESCE(t.open_key, ''), COALESCE(t.open_tier, ''), COALESCE(t.default_join, 0), |
| EXISTS(SELECT 1 FROM hidden_threads h WHERE h.thread_id = t.id AND h.user_id = m.user_id), |
| @@ -2531,10 +2541,10 @@ func (db *DB) threadsForUser(userID string) ([]Thread, error) { |
| LEFT JOIN thread_keys k ON k.thread_id = t.id AND k.user_id = m.user_id |
| WHERE m.user_id = ? |
| ORDER BY pinned DESC, |
| - COALESCE((SELECT MAX(created_at) FROM messages WHERE thread_id = t.id), t.created_at) DESC, |
| + COALESCE((SELECT MAX(created_at) FROM messages WHERE thread_id = t.id AND tombstone_of IS NULL), t.created_at) DESC, |
| -- created_at is second-resolution, so threads active in the same |
| -- second tie; message ids are monotonic, making the order stable. |
| - COALESCE((SELECT MAX(id) FROM messages WHERE thread_id = t.id), 0) DESC, |
| + COALESCE((SELECT MAX(id) FROM messages WHERE thread_id = t.id AND tombstone_of IS NULL), 0) DESC, |
| t.created_at DESC`, userID) |
| if err != nil { |
| return nil, err |
| @@ -3016,6 +3026,10 @@ type Message struct { |
| CreatedAt int64 `json:"created_at"` |
| Blob string `json:"blob"` |
| EditedAt int64 `json:"edited_at,omitempty"` |
| + // Non-zero marks this row a delete tombstone: the id of the message it |
| + // unsent (#581). Tombstone rows have an empty blob; created_at is the |
| + // deletion time. Old clients drop empty-blob rows, so this is additive. |
| + TombstoneOf int64 `json:"tombstone_of,omitempty"` |
| } |
| func (db *DB) insertMessage(threadID, blob, authorHash string) (Message, error) { |
| @@ -3065,23 +3079,58 @@ func (db *DB) editMessage(id int64, blob, authorHash string, window time.Duratio |
| // Returns whether a row was removed (false = not the author / unknown id). |
| // Reactions live inside encrypted message events, so there is nothing else |
| // to clean up here. |
| +// |
| +// In the same transaction it inserts a tombstone row (#581): fresh id, empty |
| +// blob, tombstone_of = the deleted id. The live "delete" broadcast reaches |
| +// only connected clients; the tombstone's fresh id lands above every |
| +// watermark, so clients that were offline replay the delete from the |
| +// ordinary `after:` catch-up with no new sync machinery. (Tombstones |
| +// themselves can't be deleted or edited: their author_hash is NULL, which |
| +// matches no caller.) |
| func (db *DB) deleteMessage(id int64, authorHash string) (bool, error) { |
| - res, err := db.sql.Exec("DELETE FROM messages WHERE id = ? AND author_hash = ?", id, authorHash) |
| + tx, err := db.sql.Begin() |
| if err != nil { |
| return false, err |
| } |
| - n, err := res.RowsAffected() |
| + defer tx.Rollback() |
| + var threadID sql.NullString |
| + err = tx.QueryRow("SELECT thread_id FROM messages WHERE id = ? AND author_hash = ?", id, authorHash).Scan(&threadID) |
| + if err == sql.ErrNoRows { |
| + return false, nil |
| + } |
| if err != nil { |
| return false, err |
| } |
| - return n > 0, nil |
| + // The DELETE re-checks author_hash and gates on RowsAffected, preserving the |
| + // old single-statement invariant: true means THIS call removed the row. A |
| + // raced duplicate (double-tap, retried frame) loses here and mints no |
| + // second tombstone. |
| + res, err := tx.Exec("DELETE FROM messages WHERE id = ? AND author_hash = ?", id, authorHash) |
| + if err != nil { |
| + return false, err |
| + } |
| + if n, err := res.RowsAffected(); err != nil || n == 0 { |
| + return false, err |
| + } |
| + // A legacy pre-thread_id row (NULL) is unreachable by thread reads, so no |
| + // client ever rendered it — skip the tombstone rather than insert one no |
| + // catch-up query could deliver. |
| + if threadID.Valid { |
| + if _, err := tx.Exec( |
| + "INSERT INTO messages(created_at, blob, thread_id, tombstone_of) VALUES(?, '', ?, ?)", |
| + time.Now().Unix(), threadID.String, id, |
| + ); err != nil { |
| + return false, err |
| + } |
| + } |
| + return true, tx.Commit() |
| } |
| // messagesBefore returns the `limit` messages immediately older than `before` |
| // (id < before), ascending — used to paginate history upward. |
| func (db *DB) messagesBefore(threadID string, before int64, limit int) ([]Message, error) { |
| rows, err := db.sql.Query( |
| - "SELECT id, created_at, blob, COALESCE(edited_at, 0) FROM (SELECT id, created_at, blob, edited_at FROM messages WHERE thread_id = ? AND id < ? ORDER BY id DESC LIMIT ?) ORDER BY id ASC", |
| + "SELECT id, created_at, blob, COALESCE(edited_at, 0), COALESCE(tombstone_of, 0) FROM (SELECT id, created_at, blob, edited_at, tombstone_of FROM messages WHERE thread_id = ? AND id < ? ORDER BY id DESC LIMIT ?) ORDER BY id ASC", |
| threadID, before, limit, |
| ) |
| if err != nil { |
| @@ -3091,7 +3140,7 @@ func (db *DB) messagesBefore(threadID string, before int64, limit int) ([]Messag |
| out := []Message{} |
| for rows.Next() { |
| var m Message |
| - if err := rows.Scan(&m.ID, &m.CreatedAt, &m.Blob, &m.EditedAt); err != nil { |
| + if err := rows.Scan(&m.ID, &m.CreatedAt, &m.Blob, &m.EditedAt, &m.TombstoneOf); err != nil { |
| return nil, err |
| } |
| out = append(out, m) |
| @@ -3106,12 +3155,12 @@ func (db *DB) messagesAfter(threadID string, after int64, limit int) ([]Message, |
| var err error |
| if after == 0 { |
| rows, err = db.sql.Query( |
| - "SELECT id, created_at, blob, COALESCE(edited_at, 0) FROM (SELECT id, created_at, blob, edited_at FROM messages WHERE thread_id = ? ORDER BY id DESC LIMIT ?) ORDER BY id ASC", |
| + "SELECT id, created_at, blob, COALESCE(edited_at, 0), COALESCE(tombstone_of, 0) FROM (SELECT id, created_at, blob, edited_at, tombstone_of FROM messages WHERE thread_id = ? ORDER BY id DESC LIMIT ?) ORDER BY id ASC", |
| threadID, limit, |
| ) |
| } else { |
| rows, err = db.sql.Query( |
| - "SELECT id, created_at, blob, COALESCE(edited_at, 0) FROM messages WHERE thread_id = ? AND id > ? ORDER BY id ASC LIMIT ?", |
| + "SELECT id, created_at, blob, COALESCE(edited_at, 0), COALESCE(tombstone_of, 0) FROM messages WHERE thread_id = ? AND id > ? ORDER BY id ASC LIMIT ?", |
| threadID, after, limit, |
| ) |
| } |
| @@ -3122,7 +3171,7 @@ func (db *DB) messagesAfter(threadID string, after int64, limit int) ([]Message, |
| out := []Message{} |
| for rows.Next() { |
| var m Message |
| - if err := rows.Scan(&m.ID, &m.CreatedAt, &m.Blob, &m.EditedAt); err != nil { |
| + if err := rows.Scan(&m.ID, &m.CreatedAt, &m.Blob, &m.EditedAt, &m.TombstoneOf); err != nil { |
| return nil, err |
| } |
| out = append(out, m) |
diff --git a/db_test.go b/db_test.go| index c254072..386b332 100644 |
| --- a/db_test.go |
| +++ b/db_test.go |
| @@ -51,22 +51,84 @@ func TestMessages(t *testing.T) { |
| func TestDeleteMessage(t *testing.T) { |
| db := newTestDB(t) |
| m1, _ := db.insertMessage("t1", "blob1", "author-a") |
| - db.insertMessage("t1", "blob2", "author-b") |
| + m2, _ := db.insertMessage("t1", "blob2", "author-b") |
| - // A non-author can't delete it. |
| + // A non-author can't delete it (and no tombstone appears). |
| if ok, err := db.deleteMessage(m1.ID, "author-b"); err != nil || ok { |
| t.Fatalf("non-author delete: ok=%v err=%v (want ok=false)", ok, err) |
| } |
| if all, _ := db.messagesAfter("t1", 0, 100); len(all) != 2 { |
| t.Fatalf("message wrongly deleted by a non-author") |
| } |
| - // The author can delete it; the other message survives. |
| + // The author can delete it: the ciphertext row goes, the other message |
| + // survives, and a tombstone row appears with a fresh id above everything — |
| + // so a client syncing forward from any watermark learns of the delete. |
| if ok, err := db.deleteMessage(m1.ID, "author-a"); err != nil || !ok { |
| t.Fatalf("author delete: ok=%v err=%v (want ok=true)", ok, err) |
| } |
| all, _ := db.messagesAfter("t1", 0, 100) |
| - if len(all) != 1 || all[0].Blob != "blob2" { |
| - t.Fatalf("after delete: %+v (want only blob2)", all) |
| + if len(all) != 2 || all[0].Blob != "blob2" { |
| + t.Fatalf("after delete: %+v (want blob2 + tombstone)", all) |
| + } |
| + tomb := all[1] |
| + if tomb.TombstoneOf != m1.ID || tomb.Blob != "" || tomb.ID <= m2.ID || tomb.CreatedAt == 0 { |
| + t.Fatalf("tombstone: %+v (want tombstone_of=%d, empty blob, id>%d)", tomb, m1.ID, m2.ID) |
| + } |
| + // The tombstone reaches a client that had already caught up past m2. |
| + caught, _ := db.messagesAfter("t1", m2.ID, 100) |
| + if len(caught) != 1 || caught[0].TombstoneOf != m1.ID { |
| + t.Fatalf("catch-up after %d: %+v (want just the tombstone)", m2.ID, caught) |
| + } |
| + // Tombstones themselves can't be deleted or re-tombstoned (no author). |
| + if ok, err := db.deleteMessage(tomb.ID, "author-a"); err != nil || ok { |
| + t.Fatalf("deleting a tombstone: ok=%v err=%v (want ok=false)", ok, err) |
| + } |
| + // Non-tombstone rows never carry the field on the wire (omitempty relies |
| + // on 0), and ordinary reads still return it as 0. |
| + if all[0].TombstoneOf != 0 { |
| + t.Fatalf("live message has tombstone_of=%d", all[0].TombstoneOf) |
| + } |
| +} |
| + |
| +// 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. |
| +func TestTombstoneNotThreadActivity(t *testing.T) { |
| + db := newTestDB(t) |
| + uid, err := db.createUser("Paul") |
| + if err != nil { |
| + t.Fatalf("createUser: %v", err) |
| + } |
| + t1, err := db.createThread("group", "n1", []ThreadMemberKey{{UserID: uid, Wrapped: "w1"}}) |
| + if err != nil { |
| + t.Fatalf("createThread: %v", err) |
| + } |
| + t2, err := db.createThread("group", "n2", []ThreadMemberKey{{UserID: uid, Wrapped: "w2"}}) |
| + if err != nil { |
| + t.Fatalf("createThread: %v", err) |
| + } |
| + old, _ := db.insertMessage(t1, "old", "author-a") |
| + db.insertMessage(t1, "latest-t1", "author-a") |
| + db.insertMessage(t2, "latest-t2", "author-a") |
| + |
| + // t2 has the newest message → sorts first. |
| + ts, _ := db.threadsForUser(uid) |
| + if len(ts) != 2 || ts[0].ID != t2 { |
| + t.Fatalf("pre-delete order: %+v (want t2 first)", ts) |
| + } |
| + // Deleting t1's OLD message adds a tombstone with the newest id of all — |
| + // but t2 must stay first, and t1's preview must stay its surviving message. |
| + if ok, _ := db.deleteMessage(old.ID, "author-a"); !ok { |
| + t.Fatal("delete failed") |
| + } |
| + ts, _ = db.threadsForUser(uid) |
| + if ts[0].ID != t2 { |
| + t.Fatalf("tombstone bumped thread order: %+v", ts) |
| + } |
| + for _, th := range ts { |
| + if th.ID == t1 && th.LastBlob != "latest-t1" { |
| + t.Fatalf("t1 preview: %q (want latest-t1)", th.LastBlob) |
| + } |
| } |
| } |
diff --git a/handlers_test.go b/handlers_test.go| index e156cfb..40de7f2 100644 |
| --- a/handlers_test.go |
| +++ b/handlers_test.go |
| @@ -113,6 +113,32 @@ func TestMessagesThreadScoping(t *testing.T) { |
| } |
| } |
| +// A delete's tombstone row (#581) rides the ordinary /api/messages read: |
| +// clients that were offline for the live broadcast see `tombstone_of` on a |
| +// fresh row and fold it as a delete. The wire shape is what old clients skip |
| +// (empty blob) and what the chat.json event vector pins. |
| +func TestMessagesTombstoneOnWire(t *testing.T) { |
| + ts, srv := newTestServer(t) |
| + tok, _, tid := joinedMember(t, srv, "Member") |
| + m1, _ := srv.db.insertMessage(tid, "blob1", hashToken(tok)) |
| + m2, _ := srv.db.insertMessage(tid, "blob2", hashToken(tok)) |
| + if ok, err := srv.db.deleteMessage(m1.ID, hashToken(tok)); err != nil || !ok { |
| + t.Fatalf("deleteMessage: ok=%v err=%v", ok, err) |
| + } |
| + |
| + // A client catching up from m2 (its watermark) receives just the tombstone. |
| + r, _ := http.Get(ts.URL + "/api/messages?token=" + tok + "&thread=" + tid + "&after=" + strconv.FormatInt(m2.ID, 10)) |
| + var got struct { |
| + Messages []Message `json:"messages"` |
| + } |
| + if err := json.NewDecoder(r.Body).Decode(&got); err != nil { |
| + t.Fatalf("decode: %v", err) |
| + } |
| + if len(got.Messages) != 1 || got.Messages[0].TombstoneOf != m1.ID || got.Messages[0].Blob != "" { |
| + t.Fatalf("catch-up page: %+v (want one tombstone of %d)", got.Messages, m1.ID) |
| + } |
| +} |
| + |
| func TestKeyInviteFlow(t *testing.T) { |
| ts, _ := newTestServer(t) |
diff --git a/reaper.go b/reaper.go| index b4741da..0c26645 100644 |
| --- a/reaper.go |
| +++ b/reaper.go |
| @@ -77,6 +77,10 @@ func instanceStatus(host string) (members int, setupComplete, hasContent bool, e |
| // hasContent guards against a false positive: any message or thread means the |
| // space was actually used, so it's never reaped no matter what setup_complete |
| // says. EXISTS short-circuits, so this stays O(1) even on a large DB. |
| + // NB: delete tombstones (#581) count as "content" here. That's deliberate: |
| + // this runs against OTHER instances' DB files, which may predate the |
| + // tombstone_of column (filtering on it would error), and a tombstone means |
| + // someone really used the space — erring protective is the guard's job. |
| var hasMsg, hasThread int |
| if err := conn.QueryRowContext(ctx, |
| "SELECT EXISTS(SELECT 1 FROM messages), EXISTS(SELECT 1 FROM threads)").Scan(&hasMsg, &hasThread); err != nil { |