From f63c91817ea4c53ed1fb4be4615c17468106cf08 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Mon, 3 Aug 2026 19:53:22 +1000 Subject: [PATCH 1/2] Store received messages with complete wire recipient lists MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Inbound storage truncated a message's recipients to the locally accepted ones, discarding co-recipients on other domains. That broke the participant record (hosts disagreed on a message's recipients), and worse, broke SPEC §10.3 participant validation: a reply from a co-recipient this host never recorded would be rejected as invalid. It also made faithful reconstruction of the message impossible, which SPEC §11 hash computation depends on. storeMsgDetail now stores the complete wire to/add to lists in wire order. Recipients this host accepted are marked delivered as before; recipients it rejected keep the per-recipient code it responded; and recipients on other domains carry the new local sentinel response code -2 (not our delivery) so the sender's pending queries never treat them as this host's work. attachAddToRecipients and storeMsgHeaderOnly use the same sentinel (previously code 11). Received messages additionally persist their exact wire header bytes in the new msg.wire_header column so any hash -- including future add-to batch variants -- can always be recomputed faithfully, immune to re-encode drift. Locally-authored messages leave it null. Existing rows cannot be backfilled; the discarded recipients were lost at receive time. Upgrade path for existing databases: alter table msg add column if not exists wire_header bytea; (no other schema change; the sentinel uses the existing response_code column). Co-Authored-By: Claude Fable 5 --- cmd/fmsgd/host.go | 28 +++++++++------ cmd/fmsgd/sender.go | 14 ++++++-- cmd/fmsgd/store.go | 79 +++++++++++++++++++++++++---------------- cmd/fmsgd/store_test.go | 26 ++++++++++++++ dd.sql | 8 +++-- 5 files changed, 109 insertions(+), 46 deletions(-) diff --git a/cmd/fmsgd/host.go b/cmd/fmsgd/host.go index 3cb5808..6b3f9ba 100644 --- a/cmd/fmsgd/host.go +++ b/cmd/fmsgd/host.go @@ -1367,20 +1367,20 @@ func persistAttachmentPayloads(h *FMsgHeader, dirpath string) error { return nil } -func storeAcceptedMessage(h *FMsgHeader, codes []byte, acceptedTo []FMsgAddress, acceptedAddTo []FMsgAddress, primaryFilepath string) bool { +// storeAcceptedMessage stores a received message once at least one local +// recipient accepted it. The header is stored with its COMPLETE wire +// recipient lists (never truncated to this host's recipients — §10.3 +// participant checks and §11 hash reconstruction need the message exactly as +// transmitted); localOutcome carries the per-recipient codes this host +// responded so each row records delivered/rejected/not-our-delivery. +func storeAcceptedMessage(h *FMsgHeader, codes []byte, acceptedTo []FMsgAddress, acceptedAddTo []FMsgAddress, localOutcome map[string]uint8, primaryFilepath string) bool { if len(acceptedTo) == 0 && len(acceptedAddTo) == 0 { return false } - origTo := h.To - origAddTo := h.AddTo - h.To = acceptedTo - h.AddTo = acceptedAddTo h.Filepath = primaryFilepath - if err := storeMsgDetail(h); err != nil { + if err := storeMsgDetail(h, localOutcome); err != nil { log.Printf("ERROR: storing message: %s", err) - h.To = origTo - h.AddTo = origAddTo for i := range codes { if codes[i] == RejectCodeAccept { codes[i] = RejectCodeUndisclosed @@ -1389,8 +1389,6 @@ func storeAcceptedMessage(h *FMsgHeader, codes []byte, acceptedTo []FMsgAddress, return false } - h.To = origTo - h.AddTo = origAddTo allAccepted := append(acceptedTo, acceptedAddTo...) for i := range allAccepted { if err := postMsgStatRecv(&allAccepted[i], h.Timestamp, int(h.Size)); err != nil { @@ -1506,7 +1504,15 @@ func downloadMessage(c net.Conn, r io.Reader, h *FMsgHeader, skipData bool) erro } } - stored := storeAcceptedMessage(h, codes, acceptedTo, acceptedAddTo, primaryFilepath) + // Map each local recipient to the code this host responded, so storage can + // record every wire recipient's outcome (delivered / rejected / another + // host's delivery). + localOutcome := make(map[string]uint8, len(addrs)) + for i := range addrs { + localOutcome[strings.ToLower(addrs[i].ToString())] = codes[i] + } + + stored := storeAcceptedMessage(h, codes, acceptedTo, acceptedAddTo, localOutcome, primaryFilepath) if stored { cleanupOnReturn = false } diff --git a/cmd/fmsgd/sender.go b/cmd/fmsgd/sender.go index 5c46598..579d6e4 100644 --- a/cmd/fmsgd/sender.go +++ b/cmd/fmsgd/sender.go @@ -22,8 +22,18 @@ var RetryMaxAge float64 = 86400 var PollInterval = 10 var MaxConcurrentSend = 1024 -// localResponseCodeNoResponse is stored only in the database; it is not an fmsg protocol response code. -const localResponseCodeNoResponse = -1 +// Local response codes are stored only in the database; they are not fmsg +// protocol response codes (negative so they can never collide with one). +const ( + // localResponseCodeNoResponse marks a delivery attempt that got no + // response; the row stays retryable. + localResponseCodeNoResponse = -1 + // localResponseCodeNotOurDelivery marks a recipient recorded from a + // received exchange purely for participant checks and faithful message + // reconstruction (SPEC §10.3/§11) — delivering to them is another host's + // job, so the row is never retried and never treated as pending. + localResponseCodeNotOurDelivery = -2 +) var retryableResponseCodes = []int16{ int16(localResponseCodeNoResponse), diff --git a/cmd/fmsgd/store.go b/cmd/fmsgd/store.go index af35c0e..d7bc19e 100644 --- a/cmd/fmsgd/store.go +++ b/cmd/fmsgd/store.go @@ -298,7 +298,7 @@ func attachAddToRecipients(tx *sql.Tx, msgID int64, msg *FMsgHeader) error { // This host is recording a batch it RECEIVED: delivering the batch to // other domains is the batch sender's job, not ours (SPEC §10.2 — a host // delivers iff from or add to from belongs to its domain). Rows for other - // domains are recorded with response code 11 (accept add to) so the + // domains are recorded with localResponseCodeNotOurDelivery so the // sender's pending queries never treat them as our delivery work. for _, addr := range msg.To { var delivered interface{} @@ -306,7 +306,7 @@ func attachAddToRecipients(tx *sql.Tx, msgID int64, msg *FMsgHeader) error { if addr.Domain == Domain { delivered = now } else { - code = int16(AcceptCodeAddTo) + code = int16(localResponseCodeNotOurDelivery) } if _, err := tx.Exec(`insert into msg_to (msg_id, addr, time_delivered, response_code) values ($1, $2, $3, $4) @@ -334,7 +334,7 @@ on conflict (msg_id, addr) do nothing`, msgID, addr.ToString(), delivered, code) if addr.Domain == Domain { delivered = now } else { - code = int16(AcceptCodeAddTo) + code = int16(localResponseCodeNotOurDelivery) } if _, err := tx.Exec(`insert into msg_add_to (msg_id, batch_id, addr, time_delivered, response_code) values ($1, $2, $3, $4, $5) @@ -345,7 +345,28 @@ on conflict (msg_id, addr) do nothing`, msgID, batchID, addr.ToString(), deliver return nil } -func storeMsgDetail(msg *FMsgHeader) error { +// inboundRecipientRow maps one wire recipient of a received message to its +// stored row state: recipients this host accepted are delivered; recipients +// this host rejected keep the per-recipient code it responded; recipients on +// other domains are recorded for participant checks and faithful message +// reconstruction (SPEC §10.3/§11) but are another host's delivery duty. +func inboundRecipientRow(addr FMsgAddress, localOutcome map[string]uint8, now float64) (delivered interface{}, code interface{}) { + c, ok := localOutcome[strings.ToLower(addr.ToString())] + if !ok { + return nil, int16(localResponseCodeNotOurDelivery) + } + if c == RejectCodeAccept { + return now, nil + } + return nil, int16(c) +} + +// storeMsgDetail stores a RECEIVED message: the complete wire recipient lists +// are kept in wire order — never truncated to this host's recipients — +// because §10.3's participant checks and §11's hash reconstruction both need +// the message exactly as transmitted. localOutcome maps each lower-cased +// local recipient address to the per-recipient code this host responded. +func storeMsgDetail(msg *FMsgHeader, localOutcome map[string]uint8) error { db, err := sql.Open("postgres", "") if err != nil { @@ -389,8 +410,9 @@ func storeMsgDetail(msg *FMsgHeader) error { , sha256 , psha256 , size - , filepath) -values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + , filepath + , wire_header) +values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) returning id`, msg.Version, msg.Flags&FlagNoReply != 0, @@ -403,13 +425,14 @@ returning id`, msgHash, parentHash, int(msg.Size), - msg.Filepath).Scan(&msgID) + msg.Filepath, + msg.Encode()).Scan(&msgID) if err != nil { return err } - stmt, err := tx.Prepare(`insert into msg_to (msg_id, addr, time_delivered) -values ($1, $2, $3)`) + stmt, err := tx.Prepare(`insert into msg_to (msg_id, addr, time_delivered, response_code) +values ($1, $2, $3, $4)`) if err != nil { return err } @@ -417,12 +440,8 @@ values ($1, $2, $3)`) now := timeutil.TimestampNow().Float64() for _, addr := range msg.To { - // recipients on our domain are already delivered; others are pending - var delivered interface{} - if addr.Domain == Domain { - delivered = now - } - if _, err := stmt.Exec(msgID, addr.ToString(), delivered); err != nil { + delivered, code := inboundRecipientRow(addr, localOutcome, now) + if _, err := stmt.Exec(msgID, addr.ToString(), delivered, code); err != nil { return err } } @@ -438,19 +457,16 @@ values ($1, $2, $3)`) return err } - addToStmt, err := tx.Prepare(`insert into msg_add_to (msg_id, batch_id, addr, time_delivered) -values ($1, $2, $3, $4)`) + addToStmt, err := tx.Prepare(`insert into msg_add_to (msg_id, batch_id, addr, time_delivered, response_code) +values ($1, $2, $3, $4, $5)`) if err != nil { return err } defer addToStmt.Close() for _, addr := range msg.AddTo { - var delivered interface{} - if addr.Domain == Domain { - delivered = now - } - if _, err := addToStmt.Exec(msgID, batchID, addr.ToString(), delivered); err != nil { + delivered, code := inboundRecipientRow(addr, localOutcome, now) + if _, err := addToStmt.Exec(msgID, batchID, addr.ToString(), delivered, code); err != nil { return err } } @@ -526,8 +542,9 @@ func storeMsgHeaderOnly(msg *FMsgHeader) error { , sha256 , psha256 , size - , filepath) -values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + , filepath + , wire_header) +values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) returning id`, msg.Version, msg.Flags&FlagNoReply != 0, @@ -540,19 +557,21 @@ returning id`, msgHash, parentHash, int(msg.Size), - "").Scan(&msgID) + "", + msg.Encode()).Scan(&msgID) if err != nil { return err } - // insert to recipients (for record keeping) - toStmt, err := tx.Prepare(`insert into msg_to (msg_id, addr) values ($1, $2)`) + // Record-keeping rows only: no delivery happened for anyone here, so every + // recipient is marked not-our-delivery. + toStmt, err := tx.Prepare(`insert into msg_to (msg_id, addr, response_code) values ($1, $2, $3)`) if err != nil { return err } defer toStmt.Close() for _, addr := range msg.To { - if _, err := toStmt.Exec(msgID, addr.ToString()); err != nil { + if _, err := toStmt.Exec(msgID, addr.ToString(), int16(localResponseCodeNotOurDelivery)); err != nil { return err } } @@ -568,13 +587,13 @@ returning id`, return err } - addToStmt, err := tx.Prepare(`insert into msg_add_to (msg_id, batch_id, addr) values ($1, $2, $3)`) + addToStmt, err := tx.Prepare(`insert into msg_add_to (msg_id, batch_id, addr, response_code) values ($1, $2, $3, $4)`) if err != nil { return err } defer addToStmt.Close() for _, addr := range msg.AddTo { - if _, err := addToStmt.Exec(msgID, batchID, addr.ToString()); err != nil { + if _, err := addToStmt.Exec(msgID, batchID, addr.ToString(), int16(localResponseCodeNotOurDelivery)); err != nil { return err } } diff --git a/cmd/fmsgd/store_test.go b/cmd/fmsgd/store_test.go index 98a5687..670541a 100644 --- a/cmd/fmsgd/store_test.go +++ b/cmd/fmsgd/store_test.go @@ -181,3 +181,29 @@ func TestRelationalParentHashAddToHasNoParent(t *testing.T) { t.Fatalf("new-thread relational parent = %v, want nil", got) } } + +func TestInboundRecipientRow(t *testing.T) { + now := 1234.5 + local := FMsgAddress{User: "alice", Domain: "here.example"} + rejected := FMsgAddress{User: "carol", Domain: "here.example"} + remote := FMsgAddress{User: "bob", Domain: "there.example"} + outcome := map[string]uint8{ + "@alice@here.example": RejectCodeAccept, + "@carol@here.example": RejectCodeUserFull, + } + + delivered, code := inboundRecipientRow(local, outcome, now) + if delivered != now || code != nil { + t.Fatalf("accepted local: got (%v, %v), want (%v, nil)", delivered, code, now) + } + + delivered, code = inboundRecipientRow(rejected, outcome, now) + if delivered != nil || code != int16(RejectCodeUserFull) { + t.Fatalf("rejected local: got (%v, %v), want (nil, %d)", delivered, code, RejectCodeUserFull) + } + + delivered, code = inboundRecipientRow(remote, outcome, now) + if delivered != nil || code != int16(localResponseCodeNotOurDelivery) { + t.Fatalf("remote: got (%v, %v), want (nil, %d)", delivered, code, localResponseCodeNotOurDelivery) + } +} diff --git a/dd.sql b/dd.sql index e32dcee..7a3f25b 100644 --- a/dd.sql +++ b/dd.sql @@ -20,9 +20,11 @@ create table if not exists msg ( sha256 bytea unique, psha256 bytea, size int not null, -- spec allows uint32 but we don't enforced by FMSG_MAX_MSG_SIZE - filepath text not null + filepath text not null, + wire_header bytea -- received messages: the exact wire header bytes (fields 1-13), so any hash can always be faithfully recomputed (SPEC §11); null for locally-authored messages ); create index on msg ((lower(from_addr))); +alter table msg add column if not exists wire_header bytea; -- upgrade path for databases created before this column create table if not exists msg_to ( id bigserial primary key, @@ -31,7 +33,7 @@ create table if not exists msg_to ( time_delivered double precision, -- if sending, time sending host recieved delivery confirmation, if receiving, time successfully received message time_last_attempt double precision, -- only used when sending, time of last delivery attempt if failed; otherwise null time_read double precision, -- time recipient read the message; null if unread - response_code smallint, -- only used when sending, response code of last delivery attempt if failed; otherwise null + response_code smallint, -- when sending, response code of last delivery attempt if failed; when receiving, the per-recipient code this host responded, or a negative local sentinel (-1 attempt got no response, retryable; -2 recorded from an exchange, another host's delivery) attempt_count int not null default 0, -- number of failed delivery attempts; used for exponential back-off unique (msg_id, addr) ); @@ -57,7 +59,7 @@ create table if not exists msg_add_to ( time_delivered double precision, -- if sending, time sending host recieved delivery confirmation, if receiving, time successfully received message time_last_attempt double precision, -- only used when sending, time of last delivery attempt if failed; otherwise null time_read double precision, -- time recipient read the message; null if unread - response_code smallint, -- only used when sending, response code of last delivery attempt if failed; otherwise null + response_code smallint, -- when sending, response code of last delivery attempt if failed; when receiving, the per-recipient code this host responded, or a negative local sentinel (-1 attempt got no response, retryable; -2 recorded from an exchange, another host's delivery) attempt_count int not null default 0, -- number of failed delivery attempts; used for exponential back-off unique (msg_id, addr) ); From 5d9edcdbd978716c2d791601494467b430864e77 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Mon, 3 Aug 2026 20:16:49 +1000 Subject: [PATCH 2/2] Sync SPEC.md from fmsg-spec (v0.4.1: retain stored messages in full) Co-Authored-By: Claude Fable 5 --- SPEC.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/SPEC.md b/SPEC.md index da45461..7ebcf1a 100644 --- a/SPEC.md +++ b/SPEC.md @@ -270,6 +270,8 @@ A message is verified as stored iff: - A SHA-256 digest matches a previously accepted message (code 200 or 11). - That message currently exists and is retrievable. +A host MUST retain each stored message in full and exactly as transmitted — including the complete _to_ and _add to_ recipient lists, not only recipients on its own domain — so the message hash can always be faithfully recomputed and participant checks (§10.3 step 7) evaluate against the true participant set. + For accept-add-to (code 11) messages, the hash is computed by combining the add-to message header with the original message's data and attachment data. Each add-to batch produces a distinct hash. Only the exact batch that had an accepted response (200 or 11) matches.