Skip to content

Identify add-to batches by batch message hash #35

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

Open
markmnl wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Identify add-to batches by batch message hash #35

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
24 changes: 17 additions & 7 deletions cmd/fmsgd/host.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -607,8 +607,24 @@ func handleAddToPath(c net.Conn, h *FMsgHeader) (*FMsgHeader, error) {
return h, fmt.Errorf("add-to: time travel detected (parent time %f, current %f)", parentMsg.Timestamp, h.Timestamp)
}

// Batch identity is the batch message hash — this add-to header combined
// with the stored parent's payload (SPEC §11) — so map the parent's data
// in and compute it before the duplicate check.
h.Filepath = parentMsg.Filepath
for i := range h.Attachments {
if i < len(parentMsg.Attachments) {
h.Attachments[i].Filepath = parentMsg.Attachments[i].Filepath
}
}
batchHash, err := h.GetMessageHash()
if err != nil {
return h, err
}

// A batch this host already recorded is a duplicate (SPEC §10.4 step 1).
recorded, err := addToBatchRecorded(parentID, h.AddTo)
// The same addresses re-issued at a new time hash differently and are a
// distinct batch — a new sibling branch — not a duplicate (SPEC §12).
recorded, err := addToBatchRecorded(parentID, batchHash)
if err != nil {
return h, err
}
Expand All @@ -624,12 +640,6 @@ func handleAddToPath(c net.Conn, h *FMsgHeader) (*FMsgHeader, error) {
return h, nil
}

h.Filepath = parentMsg.Filepath
for i := range h.Attachments {
if i < len(parentMsg.Attachments) {
h.Attachments[i].Filepath = parentMsg.Attachments[i].Filepath
}
}
h.InitialResponseCode = AcceptCodeAddTo
return h, nil
}
Expand Down
74 changes: 40 additions & 34 deletions cmd/fmsgd/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,14 +286,13 @@ func existingMsgIDForAddTo(tx *sql.Tx, msg *FMsgHeader, msgHash []byte) (int64,
return id, err
}

// addToBatchRecorded reports whether an incoming add-to batch carries nothing
// this host has not already recorded against stored message msgID: every
// address is unique per message across batches (msg_add_to unique (msg_id,
// addr)), so when every address in the incoming batch is already attached the
// delivery is a re-send of a recorded batch and is a duplicate (code 10,
// SPEC §10.4 step 1).
func addToBatchRecorded(msgID int64, addTo []FMsgAddress) (bool, error) {
if len(addTo) == 0 {
// addToBatchRecorded reports whether this host has already recorded an add-to
// batch with this batch message hash against stored message msgID. Batch
// identity IS the batch message hash, which covers time (SPEC §11): the same
// addresses re-issued at a new time hash differently and are a distinct
// batch, not a duplicate (SPEC §12).
func addToBatchRecorded(msgID int64, batchHash []byte) (bool, error) {
if len(batchHash) == 0 {
return false, nil
}
db, err := sql.Open("postgres", "")
Expand All @@ -302,29 +301,22 @@ func addToBatchRecorded(msgID int64, addTo []FMsgAddress) (bool, error) {
}
defer db.Close()

for i := range addTo {
var exists bool
err = db.QueryRow(`SELECT EXISTS (
SELECT 1 FROM msg_add_to WHERE msg_id = $1 AND lower(addr) = $2
)`, msgID, strings.ToLower(addTo[i].ToString())).Scan(&exists)
if err != nil {
return false, err
}
if !exists {
return false, nil
}
}
return true, nil
}

// insertAddToBatch records one add-to delivery as a batch (its sender and the
// time this host recorded it) and returns the new batch id. Recipients carried
// by the delivery are linked to this batch so readers can reconstruct who added
// which recipients and when (SPEC §12).
func insertAddToBatch(tx *sql.Tx, msgID int64, addToFrom string, now float64) (int64, error) {
var exists bool
err = db.QueryRow(`SELECT EXISTS (
SELECT 1 FROM msg_add_to_batch WHERE msg_id = $1 AND sha256 = $2
)`, msgID, batchHash).Scan(&exists)
return exists, err
}

// insertAddToBatch records one add-to delivery as a batch (its sender, the
// time this host recorded it, and its identifying batch message hash, SPEC
// §11) and returns the new batch id. Recipients carried by the delivery are
// linked to this batch so readers can reconstruct who added which recipients
// and when (SPEC §12).
func insertAddToBatch(tx *sql.Tx, msgID int64, addToFrom string, now float64, batchHash []byte) (int64, error) {
var batchID int64
err := tx.QueryRow(`insert into msg_add_to_batch (msg_id, add_to_from, time_added)
values ($1, $2, $3) returning id`, msgID, addToFrom, now).Scan(&batchID)
err := tx.QueryRow(`insert into msg_add_to_batch (msg_id, add_to_from, time_added, sha256)
values ($1, $2, $3, $4) returning id`, msgID, addToFrom, now, batchHash).Scan(&batchID)
return batchID, err
}

Expand Down Expand Up @@ -365,7 +357,14 @@ on conflict (msg_id, addr) do nothing`, msgID, addr.ToString(), delivered, code)
if msg.AddToFrom != nil {
addToFrom = msg.AddToFrom.ToString()
}
batchID, err := insertAddToBatch(tx, msgID, addToFrom, now)
// Cached since the header exchange computed it against the stored
// parent's payload (handleAddToPath); it is the batch's identity (SPEC
// §11) and what duplicate detection compares against.
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)
if err != nil {
return err
}
Expand All @@ -380,7 +379,7 @@ on conflict (msg_id, addr) do nothing`, msgID, addr.ToString(), delivered, code)
}
if _, err := tx.Exec(`insert into msg_add_to (msg_id, batch_id, addr, time_delivered, response_code)
values ($1, $2, $3, $4, $5)
on conflict (msg_id, addr) do nothing`, msgID, batchID, addr.ToString(), delivered, code); err != nil {
on conflict (batch_id, addr) do nothing`, msgID, batchID, addr.ToString(), delivered, code); err != nil {
return err
}
}
Expand Down Expand Up @@ -494,7 +493,13 @@ values ($1, $2, $3, $4)`)
if msg.AddToFrom != nil {
addToFrom = msg.AddToFrom.ToString()
}
batchID, err := insertAddToBatch(tx, msgID, addToFrom, now)
// Cached from download verification: the wire hash of an add-to
// message is its batch hash, the batch's identity (SPEC §11).
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)
if err != nil {
return err
}
Expand Down Expand Up @@ -624,7 +629,8 @@ returning id`,
if msg.AddToFrom != nil {
addToFrom = msg.AddToFrom.ToString()
}
batchID, err := insertAddToBatch(tx, msgID, addToFrom, timeutil.TimestampNow().Float64())
// No stored parent payload here, so the batch hash cannot be computed.
batchID, err := insertAddToBatch(tx, msgID, addToFrom, timeutil.TimestampNow().Float64(), nil)
if err != nil {
return err
}
Expand Down
17 changes: 14 additions & 3 deletions dd.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,13 +55,19 @@ create index if not exists msg_to_lower_idx on msg_to ((lower(addr)));
-- Each add-to delivery for a shared message is one batch: a single sender
-- (add_to_from) added a set of recipients at a point in time. Storing batches
-- separately lets readers reconstruct who added which recipients and when,
-- which a single flat recipient list cannot preserve (SPEC §12).
-- which a single flat recipient list cannot preserve (SPEC §12). A batch's
-- identity is its message hash (sha256), which covers the batch's time: the
-- same addresses re-issued at a new time are a distinct batch, not a
-- duplicate (SPEC §11/§12). sha256 is null for rows recorded before this
-- column existed and for locally originated batches not yet hashed.
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, -- when this host recorded the batch
sha256 bytea -- batch message hash: the batch's identity (SPEC §11)
);
alter table msg_add_to_batch add column if not exists sha256 bytea;
create index if not exists msg_add_to_batch_msg_id_idx on msg_add_to_batch (msg_id);

create table if not exists msg_add_to (
Expand All @@ -74,8 +80,13 @@ create table if not exists msg_add_to (
time_read double precision, -- time recipient read the message; null if unread
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)
unique (batch_id, addr)
);
-- An address is unique within a batch, not across batches: distinct batches
-- may re-add the same address (each batch is its own sibling branch, SPEC
-- §12). Migrate existing databases off the old per-message constraint.
alter table msg_add_to drop constraint if exists msg_add_to_msg_id_addr_key;
create unique index if not exists msg_add_to_batch_id_addr_key on msg_add_to (batch_id, addr);
create index if not exists msg_add_to_lower_idx on msg_add_to ((lower(addr)));
create index if not exists msg_add_to_batch_id_idx on msg_add_to (batch_id);

Expand Down
Loading