Skip to content

Store received messages with complete wire recipient lists #29

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
markmnl merged 2 commits into from
Aug 3, 2026
Merged

Store received messages with complete wire recipient lists #29

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f63c918
Store received messages with complete wire recipient lists
markmnl Aug 3, 2026
5d9edcd
Sync SPEC.md from fmsg-spec (v0.4.1: retain stored messages in full)
markmnl Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions SPEC.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
28 changes: 17 additions & 11 deletions cmd/fmsgd/host.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 {
Expand Down Expand Up @@ -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
}
Expand Down
14 changes: 12 additions & 2 deletions cmd/fmsgd/sender.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
79 changes: 49 additions & 30 deletions cmd/fmsgd/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,15 +298,15 @@ 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{}
var code interface{}
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)
Expand Down Expand Up @@ -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)
Expand All @@ -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 {
Expand Down Expand Up @@ -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,
Expand All @@ -403,26 +425,23 @@ 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
}
defer stmt.Close()

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
}
}
Expand All @@ -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
}
}
Expand Down Expand Up @@ -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,
Expand All @@ -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
}
}
Expand All @@ -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
}
}
Expand Down
26 changes: 26 additions & 0 deletions cmd/fmsgd/store_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}
8 changes: 5 additions & 3 deletions dd.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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)
);
Expand All @@ -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)
);
Expand Down
Loading