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