Resolve reply parents referencing add-to batch hashes

cmd/fmsgd/host.go

@@ -583,6 +583,9 @@ func handleAddToPath(c net.Conn, h *FMsgHeader) (*FMsgHeader, error) {
 	addToHasOurDomain := hasDomainRecipient(h.AddTo, Domain)
 	hasLocalRecipient := addToHasOurDomain || hasDomainRecipient(h.To, Domain)
 
+	// Deliberately resolves canonical message hashes only: batches do not
+	// chain, so an add-to whose pid is another batch's hash must not resolve
+	// (SPEC §12).
 	parentID, err := lookupMsgIdByHash(h.Pid)
 	if err != nil {
 		return h, err
@@ -653,22 +656,24 @@ func validatePidReplyPath(c net.Conn, h *FMsgHeader) error {
 	if err != nil {
 		return err
 	}
-	if parentID == 0 {
-		if err := sendCode(c, RejectCodeParentNotFound); err != nil {
-			return err
-		}
-		return fmt.Errorf("pid reply: parent not found for pid %s", hex.EncodeToString(h.Pid))
-	}
 
-	parentMsg, err := getMsgByID(parentID)
+	var parentMsg *FMsgHeader
+	if parentID != 0 {
+		parentMsg, err = getMsgByID(parentID)
+	} else {
+		// A reply may reference an add-to batch message via pid (SPEC §12);
+		// its wire form is reconstructed from the stored shared message and
+		// batch fields (SPEC §11).
+		parentMsg, err = getMsgByBatchHash(h.Pid)
+	}
 	if err != nil {
 		return err
 	}
 	if parentMsg == nil {
 		if err := sendCode(c, RejectCodeParentNotFound); err != nil {
 			return err
 		}
-		return fmt.Errorf("pid reply: parent message not found by ID %d", parentID)
+		return fmt.Errorf("pid reply: parent not found for pid %s", hex.EncodeToString(h.Pid))
 	}
 	if !isMessageRetrievable(parentMsg) {
 		if err := sendCode(c, RejectCodeParentNotFound); err != nil {

cmd/fmsgd/sender.go

@@ -491,6 +491,19 @@ func deliverMessage(target pendingTarget) {
 		}
 		h := m.addToHeader(b, sharedHash)
 		d.applyTo(h)
+		// Persist the batch hash — the batch's identity (SPEC §11) — once,
+		// so replies referencing this batch resolve at this host too, which
+		// must verify messages it sent, not only ones it received. Cached on
+		// h, so the challenge response reuses this computation.
+		batchHash, err := h.GetMessageHash()
+		if err != nil {
+			log.Printf("ERROR: sender: computing batch hash for batch %d of msg %d: %s", b.ID, target.MsgID, err)
+			continue
+		}
+		if err := ensureBatchHash(db, b.ID, batchHash); err != nil {
+			log.Printf("ERROR: sender: %s", err)
+			continue
+		}
 		deliverUnit(db, target, h, "msg_add_to", b.ID)
 	}
 }
@@ -515,6 +528,17 @@ func markLocalDelivered(target pendingTarget) {
 	}
 }
 
+// ensureBatchHash persists an add-to batch's message hash when not yet stored.
+// Like ensureSharedHash for the canonical hash, this is what lets replies that
+// reference the batch via pid resolve on the host that originated the batch
+// (SPEC §11: a host verifies messages it sent, not only ones it received).
+func ensureBatchHash(db *sql.DB, batchID int64, batchHash []byte) error {
+	if _, err := db.Exec(`UPDATE msg_add_to_batch SET sha256 = $1 WHERE id = $2 AND sha256 IS NULL`, batchHash, batchID); err != nil {
+		return fmt.Errorf("storing sha256 for add-to batch %d: %w", batchID, err)
+	}
+	return nil
+}
+
 // ensureSharedHash persists the message's canonical hash when not yet stored and
 // resolves any pending child (reply/add-to) links that reference it.
 func ensureSharedHash(db *sql.DB, msgID int64, sharedHash []byte) error {

cmd/fmsgd/store.go

@@ -86,6 +86,7 @@ func threadHasFromDomain(hash []byte, domain string) (bool, error) {
 			SELECT id, from_addr, pid, 1 AS depth, ARRAY[id] AS seen
 			FROM msg
 			WHERE sha256 = $1
+			   OR id IN (SELECT msg_id FROM msg_add_to_batch WHERE sha256 = $1)
 			UNION ALL
 			SELECT m.id, m.from_addr, m.pid, t.depth + 1, t.seen || m.id
 			FROM msg m
@@ -155,8 +156,15 @@ type txParentLinkStore struct {
 }
 
 func (s txParentLinkStore) lookupParentID(parentHash []byte) (int64, error) {
+	// A reply's pid may reference a message's canonical hash or one of its
+	// add-to batch hashes (SPEC §12); either way the relational parent is the
+	// shared message row.
 	var id int64
-	err := s.tx.QueryRow("SELECT id FROM msg WHERE sha256 = $1", parentHash).Scan(&id)
+	err := s.tx.QueryRow(`
+		SELECT id FROM msg WHERE sha256 = $1
+		UNION ALL
+		SELECT msg_id FROM msg_add_to_batch WHERE sha256 = $1
+		LIMIT 1`, parentHash).Scan(&id)
 	if err == sql.ErrNoRows {
 		return 0, nil
 	}
@@ -269,6 +277,57 @@ func getMsgByID(msgID int64) (*FMsgHeader, error) {
 	return h, nil
 }
 
+// getMsgByBatchHash reconstructs the wire form of an add-to batch message
+// identified by its batch hash, or nil when no such batch is recorded. A reply
+// may reference a batch via pid, and per SPEC §11 the batch message is
+// reconstructible from what the host already holds — the stored shared message
+// plus the batch's sender, recipients and wire time — even though the batch's
+// data was never downloaded again.
+func getMsgByBatchHash(batchHash []byte) (*FMsgHeader, error) {
+	if len(batchHash) == 0 {
+		return nil, nil
+	}
+	db, err := sql.Open("postgres", "")
+	if err != nil {
+		return nil, err
+	}
+	defer db.Close()
+
+	tx, err := db.Begin()
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	var msgID, batchID int64
+	err = tx.QueryRow(`SELECT msg_id, id FROM msg_add_to_batch WHERE sha256 = $1`, batchHash).Scan(&msgID, &batchID)
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	m, err := loadMsgFields(tx, msgID)
+	if err != nil {
+		return nil, err
+	}
+	batches, err := loadAddToBatches(tx, msgID)
+	if err != nil {
+		return nil, err
+	}
+	for i := range batches {
+		if batches[i].ID == batchID {
+			sharedHash, err := m.sharedHash()
+			if err != nil {
+				return nil, err
+			}
+			return m.addToHeader(batches[i], sharedHash), nil
+		}
+	}
+	return nil, fmt.Errorf("add-to batch %d missing for msg %d", batchID, msgID)
+}
+
 // existingMsgIDForAddTo returns the id of an already-stored message row whose
 // canonical sha256 matches msgHash, for an add-to delivery. It returns 0 when
 // the message is not an add-to message or no such row exists, so the caller
@@ -364,7 +423,10 @@ on conflict (msg_id, addr) do nothing`, msgID, addr.ToString(), delivered, code)
 	if err != nil {
 		return fmt.Errorf("compute add-to batch hash: %w", err)
 	}
-	batchID, err := insertAddToBatch(tx, msgID, addToFrom, now, batchHash)
+	// The batch's WIRE time is stored, not the local record time: the batch
+	// message must be reconstructible exactly as transmitted so its hash can
+	// be recomputed and replies referencing it verified (SPEC §11).
+	batchID, err := insertAddToBatch(tx, msgID, addToFrom, msg.Timestamp, batchHash)
 	if err != nil {
 		return err
 	}
@@ -494,12 +556,14 @@ values ($1, $2, $3, $4)`)
 			addToFrom = msg.AddToFrom.ToString()
 		}
 		// Cached from download verification: the wire hash of an add-to
-		// message is its batch hash, the batch's identity (SPEC §11).
+		// message is its batch hash, the batch's identity (SPEC §11). The
+		// batch's wire time is stored so the batch message stays
+		// reconstructible exactly as transmitted.
 		batchHash, err := msg.GetMessageHash()
 		if err != nil {
 			return fmt.Errorf("compute add-to batch hash: %w", err)
 		}
-		batchID, err := insertAddToBatch(tx, msgID, addToFrom, now, batchHash)
+		batchID, err := insertAddToBatch(tx, msgID, addToFrom, msg.Timestamp, batchHash)
 		if err != nil {
 			return err
 		}

dd.sql

@@ -64,7 +64,7 @@ create table if not exists msg_add_to_batch (
 	id				bigserial			primary key,
 	msg_id			bigint				not null references msg (id),
 	add_to_from		varchar(255)		not null,           -- sender that added this batch's recipients
-	time_added		double precision	not null,           -- when this host recorded the batch
+	time_added		double precision	not null,           -- the batch message's wire time field (for locally originated batches, when the batch was created)
 	sha256			bytea                                   -- batch message hash: the batch's identity (SPEC §11)
 );
 alter table msg_add_to_batch add column if not exists sha256 bytea;
@@ -134,7 +134,14 @@ begin
     if NEW.psha256 is null or octet_length(NEW.psha256) = 0 then
         NEW.psha256 = parent_sha256;
     elsif NEW.psha256 <> parent_sha256 then
-        raise exception 'psha256 does not match parent message % sha256', NEW.pid;
+        -- a reply may reference one of the parent's add-to batch messages by
+        -- its batch hash (SPEC §12); the relational parent is the shared row
+        if not exists (
+            select 1 from msg_add_to_batch b
+            where b.msg_id = NEW.pid and b.sha256 = NEW.psha256
+        ) then
+            raise exception 'psha256 does not match parent message % sha256 or any of its add-to batch hashes', NEW.pid;
+        end if;
     end if;
 
     return NEW;