diff --git a/CHANGELOG.md b/CHANGELOG.md index 69e18d7..90ef469 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- The inbox byte cap no longer deletes the whole inbox. With the default + config (`InboxMaxBytes == 0`, meaning 256 MiB) the evictor compared the + inbox size against the raw value 0, so the first time the inbox passed + 256 MiB every stored message was removed, including replies a waiting + `pilotctl send-message --wait` was about to read. Both eviction paths now + use the effective cap, evict oldest-first down to 90% of it, and never + evict for a message that cannot fit at all. +- The byte cap is checked against a running total instead of listing and + stat'ing the whole inbox on every incoming message. A message that another + connection is still writing (admitted, but not yet acknowledged) is never + evicted and never counted twice. +- When the byte cap is full the inbox now evicts the oldest messages to make + room (as `InboxMaxBytes` was documented to do) instead of rejecting the + new message. + +### Added + +- Optional request/reply correlation: `Frame.MessageID` and `Frame.ReplyTo`, + carried on the wire in a new `TypeTagged` (10) wrapper and recorded in + inbox JSON and `message.received` / `file.received` events as + `message_id` / `reply_to`. Frames without them are byte-for-byte + unchanged. `NewMessageID` and `ValidMessageID` helpers. +- `Client.Send`, which waits for the ACK and delivers a tagged frame + untagged, with `SendResult.Tagged == false`, when the tagged form cannot + get through. That happens when the receiver predates `TypeTagged`: Send + recognises both the `ERR UNKNOWN(10) ...` answer of dataexchange v0.2.2 and + later and the `ACK UNKNOWN(10) bytes` answer of v0.2.1 and older (every + stable daemon through v1.13.9), which drop the frame while acknowledging + it. It also happens when the header would push the frame over the max + frame size. +- `MaxTaggedOverhead` (293 bytes) and `ErrTaggedFrameTooLarge`: the tagged + header counts toward `MaxFrameSize`, and `WriteFrame` refuses a tagged + frame that exceeds it, writing nothing, rather than send a frame the + receiver would drop the connection over. +- Receiver-side duplicate suppression: an identical re-delivery of a stored + frame that carried a `MessageID` is acknowledged (`... (duplicate)`) but + not stored twice (`ServiceConfig.DedupeWindow`, default 10 min). + `ServiceConfig.DedupeContentWindow` (off by default) extends this to + frames without a `MessageID`. + ### Security / hardening - Lower the default per-frame cap (`DefaultMaxFrameSize`) from 1 GiB to diff --git a/README.md b/README.md index f5b20c5..1f3212a 100644 --- a/README.md +++ b/README.md @@ -30,11 +30,14 @@ rt.Register(dataexchange.NewService(dataexchange.ServiceConfig{})) | File | What it does | |---|---| -| `dataexchange.go` | Wire format: `Frame`, `WriteFrame`, `ReadFrame`, `TraceFrame`, `TypeText/Binary/JSON/File/Trace/Governed`, `TypeName`. | -| `client.go` | `Client` — `Dial` and send helpers. | +| `dataexchange.go` | Wire format: `Frame`, `WriteFrame`, `ReadFrame`, `TraceFrame`, `TypeText/Binary/JSON/File/Trace/Governed/Tagged`, `TypeName`. | +| `tagged.go` | Optional request/reply correlation on the wire: `Frame.MessageID` / `Frame.ReplyTo` (`TypeTagged`), `NewMessageID`, `ValidMessageID`. | +| `client.go` | `Client` — `Dial`, `Send` (waits for the ACK, falls back for older receivers) and send helpers. | | `governed.go` | Signed decision envelope, receiver-side verifier, and enforceable transport constraints. | | `server.go` | `Server` — accept loop and handler dispatch. | | `service.go` | `*Service` — `coreapi.Service` adapter. Build tag `!no_dataexchange`. | +| `inbox_budget.go` | Inbox caps: running byte total, oldest-first eviction to 90% of the byte cap, file-count cap. | +| `dedupe.go` | Receiver-side suppression of identical re-deliveries. | | `service_disabled.go` | Stub `*Service` for `-tags no_dataexchange` builds. | ## Wire format @@ -45,9 +48,126 @@ rt.Register(dataexchange.NewService(dataexchange.ServiceConfig{})) For `TypeFile` the payload is prefixed with `[2-byte name length][name bytes]`. For `TypeTrace` the payload is `[4-byte inner_type][8-byte sent_at_ns][inner payload]`. +For `TypeTagged` (10) the payload is +`[2-byte header_len][header: JSON object][4-byte inner_type][inner payload]`; +see [Request/reply correlation](#requestreply-correlation-for-service-authors). Max frame size: 64 MiB by default (configurable at process start with -`PILOT_DATAEXCHANGE_MAX_FRAME` within its documented safe range). +`PILOT_DATAEXCHANGE_MAX_FRAME` within its documented safe range). The limit +applies to the whole payload after the 8-byte header, so for `TypeTagged` it +includes the correlation header (see below). + +## Inbox + +Text, JSON and binary messages are written to `~/.pilot/inbox/` as one JSON +file each: + +```json +{"type":"TEXT","from":"0:0000.0002.BBE4","bytes":5,"received_at":"2026-09-24T10:00:00.123456789Z", + "data":"hello","data_encoding":"utf8","message_id":"4f1c…","reply_to":"9b2e…"} +``` + +`message_id` and `reply_to` appear only when the sender supplied them (see +below). Binary or non-UTF-8 payloads are stored as `data_b64` with +`"data_encoding":"base64"`. + +Two caps bound the inbox, and both evict the **oldest** messages first: + +- `ServiceConfig.InboxMaxFiles` (default 10000), checked every 64 messages. +- `ServiceConfig.InboxMaxBytes` (default 256 MiB), checked before every write + against a running byte total (no directory scan per message). When a new + message would not fit, the oldest messages are removed until the inbox plus + the new message is at or below 90% of the cap, so recent replies survive + and the next messages do not each trigger another eviction. A message + that another connection is still writing is never evicted. A single + message larger than the cap is rejected without evicting anything. A + negative value disables the byte cap. + +## Request/reply correlation (for service authors) + +A data-exchange message is fire-and-forget: the sender gets an ACK that the +peer stored it, and any answer arrives later as a separate message in the +sender's inbox. To let a requester tell *which* request an answer belongs to, +a frame can carry two optional fields: + +| Field (Go) | Inbox / event key | Meaning | +|---|---|---| +| `Frame.MessageID` | `message_id` | An ID the sender chose for this message. Use a fresh random value per message (`NewMessageID()`, 32 hex chars). | +| `Frame.ReplyTo` | `reply_to` | The `message_id` of the message this one answers. | + +IDs are 1–128 characters of `A-Z a-z 0-9 - _ . :`. They are correlation +metadata only: they are not authenticated (a governed frame's signature does +not cover them), so never use them for authorization. + +**If you run a service** (an agent that answers requests): when a request's +inbox record (or `message.received` event) has a `message_id`, send your +answer with `ReplyTo` set to that value. If you answer in several messages, +set it on each. Give each answer its own `MessageID` too if you want the +requester's daemon to drop duplicate copies of it (for example when you +retry a send or it arrives over two network paths). Requests without a +`message_id` come from older clients: answer them as before. + +```go +c, err := dataexchange.Dial(drv, requester) +if err != nil { return err } +defer c.Close() +res, err := c.Send(&dataexchange.Frame{ + Type: dataexchange.TypeJSON, + Payload: answer, + MessageID: dataexchange.NewMessageID(), + ReplyTo: request.MessageID, // the request's "message_id" +}) +``` + +**If you send requests**: generate a `MessageID`, send with `Client.Send`, +then wait for an inbox record from that peer whose `reply_to` equals it. +Services that predate this field answer without `reply_to`; for those, fall +back to matching on the sender alone. + +**Compatibility.** Frames without either field use the original wire format, +byte for byte. A frame with either field travels as `TypeTagged`. A receiver +that predates it stores nothing and answers in one of two ways: + +- dataexchange v0.2.1 and older (every stable pilot daemon through v1.13.9) + has no case for an unknown frame type. It drops the frame and still + answers `ACK UNKNOWN(10) bytes`. +- dataexchange v0.2.2 and later answers `ERR UNKNOWN(10) save failed: ...`. + +`Client.Send` recognises both answers, re-sends the same frame without the +fields on the same connection, and reports `SendResult.Tagged == false`, so a +new sender can always talk to an old receiver and the message is stored +exactly once. A raw `WriteFrame` of a tagged frame does not fall back: +against a v0.2.1-or-older receiver the message is silently lost, so use +`Client.Send`. + +**Size.** The tagged header adds up to `MaxTaggedOverhead` (293) bytes, and +the whole tagged frame must fit in the max frame size. A payload within 293 +bytes of the limit may fit only untagged. `Client.Send` then sends it +untagged (`SendResult.Tagged == false`), and a raw `WriteFrame` returns +`ErrTaggedFrameTooLarge` without writing anything. + +Whenever `Tagged` is false, the reply cannot carry `reply_to`; match it on +the sender alone. The fields can accompany text, JSON, binary, file, +governed and trace frames. On `TypeFileStream` frames they are ignored. + +Wire layout of a tagged frame, for other languages: + +``` +[4-byte type = 10][4-byte length][payload] +payload = [2-byte header_len, 1..1024][header_len bytes: JSON object] + [4-byte inner_type][inner payload, exactly as the inner type sends it] +header = {"message_id":"…","reply_to":"…"} (both optional; unknown keys are ignored) +length = 2 + header_len + 4 + len(inner payload), at most the max frame size +``` + +**Duplicate deliveries.** The receiver remembers every stored frame that +carried a `MessageID` for `ServiceConfig.DedupeWindow` (default 10 minutes). +An identical re-delivery — same sender, `MessageID`, `ReplyTo`, type, +filename and bytes — is acknowledged with the original ACK plus +` (duplicate)` (`SendResult.Duplicate`) and is not stored again. Frames +without a `MessageID` are always stored, as before, unless the operator sets +`ServiceConfig.DedupeContentWindow` (off by default) to also drop +byte-identical copies from the same sender inside a short window. ## Governed delivery diff --git a/client.go b/client.go index 69db791..0ab2659 100644 --- a/client.go +++ b/client.go @@ -3,8 +3,10 @@ package dataexchange import ( + "errors" "fmt" "io" + "strings" "time" "github.com/pilot-protocol/common/decision" @@ -26,6 +28,135 @@ func Dial(d *driver.Driver, addr protocol.Addr) (*Client, error) { return &Client{conn: conn}, nil } +// ErrRejected is wrapped by the error Client.Send returns when the receiver +// answered with an "ERR ..." acknowledgement instead of storing the frame. +var ErrRejected = errors.New("dataexchange: receiver rejected frame") + +// SendResult reports what the receiver did with a frame sent by Client.Send. +type SendResult struct { + // Ack is the receiver's acknowledgement frame, normally a TypeText + // "ACK bytes" (or "ERR ..." on rejection). + Ack *Frame + // Tagged is true when the frame's MessageID/ReplyTo reached the + // receiver. It is false when the frame carried neither, and also when + // Send delivered the frame without them: because the receiver predates + // TypeTagged, or because the tagged header would have pushed the frame + // over MaxFrameSize (see ErrTaggedFrameTooLarge). A requester that sees + // Tagged == false cannot expect a reply carrying reply_to, and should + // match the answer on the sender alone. + Tagged bool + // Duplicate is true when the receiver recognised the frame as an + // identical re-delivery of one it had already stored, and did not store + // it a second time. + Duplicate bool +} + +// duplicateAckSuffix is appended to the ACK of a frame the receiver +// suppressed as a duplicate. The ACK still starts with "ACK ", so senders +// that only check the prefix keep treating it as success. +const duplicateAckSuffix = " (duplicate)" + +// untaggedReceiverAckPrefixes are the two ways a receiver that predates +// TypeTagged answers a tagged frame. Neither kind stores anything: +// +// - dataexchange v0.1.0 through v0.2.1 (and the pseudo-versions between +// them) have no case for an unknown frame type. The frame matches +// nothing, no save is attempted, and the receiver answers +// "ACK UNKNOWN(10) bytes": a success ACK for a frame it dropped. +// - dataexchange v0.2.2 and later, before TypeTagged, reject an unknown +// type and answer "ERR UNKNOWN(10) save failed: unsupported frame type 10". +// A RequireGoverned receiver answers "ERR UNKNOWN(10) save failed: +// unsigned legacy frame ..." instead, also without storing it. +// +// A receiver that knows TypeTagged unwraps the frame in ReadFrame and +// acknowledges the inner type (which can never be TypeTagged), so its ACK +// never starts with either prefix. +var untaggedReceiverAckPrefixes = [...]string{ + fmt.Sprintf("ACK UNKNOWN(%d) ", TypeTagged), + fmt.Sprintf("ERR UNKNOWN(%d) ", TypeTagged), +} + +// receiverPredatesTagged reports whether ack is a pre-TypeTagged receiver's +// answer to a tagged frame, which means the frame was not stored. +func receiverPredatesTagged(ack *Frame) bool { + if ack.Type != TypeText { + return false + } + for _, prefix := range untaggedReceiverAckPrefixes { + if strings.HasPrefix(string(ack.Payload), prefix) { + return true + } + } + return false +} + +// Send writes frame, waits for the receiver's acknowledgement and reports +// the outcome. It is the preferred way to send a frame that carries a +// MessageID or ReplyTo, because it delivers the frame untagged (and reports +// SendResult.Tagged == false) in the two cases where the tagged form cannot +// get through: +// +// - The receiver predates TypeTagged. Send recognises its answer (see +// untaggedReceiverAckPrefixes) and re-sends the same frame without the +// metadata on the same connection. Such a receiver never stores the +// tagged copy, so the message is stored exactly once. +// - The tagged header would push the frame over MaxFrameSize. Send writes +// the frame untagged in the first place instead of sending a frame the +// receiver would drop the connection over. +// +// If the receiver answers "ERR ...", Send returns the result together with +// an error wrapping ErrRejected. Send reads from the connection, so do not +// use it concurrently with Recv on the same Client. +func (c *Client) Send(frame *Frame) (*SendResult, error) { + return sendAndAwaitAck(c.conn, frame) +} + +// withoutTags returns a copy of frame without MessageID and ReplyTo. +func withoutTags(frame *Frame) *Frame { + plain := *frame + plain.MessageID, plain.ReplyTo = "", "" + return &plain +} + +func sendAndAwaitAck(rw io.ReadWriter, frame *Frame) (*SendResult, error) { + if frame == nil { + return nil, fmt.Errorf("dataexchange: frame is required") + } + tagged := frame.MessageID != "" || frame.ReplyTo != "" + err := WriteFrame(rw, frame) + if tagged && errors.Is(err, ErrTaggedFrameTooLarge) { + // WriteFrame wrote nothing. The frame fits only without its header. + tagged = false + err = WriteFrame(rw, withoutTags(frame)) + } + if err != nil { + return nil, err + } + ack, err := ReadFrame(rw) + if err != nil { + return nil, fmt.Errorf("dataexchange: read ack: %w", err) + } + if tagged && receiverPredatesTagged(ack) { + if err := WriteFrame(rw, withoutTags(frame)); err != nil { + return nil, err + } + if ack, err = ReadFrame(rw); err != nil { + return nil, fmt.Errorf("dataexchange: read ack: %w", err) + } + tagged = false + } + text := string(ack.Payload) + result := &SendResult{ + Ack: ack, + Tagged: tagged, + Duplicate: ack.Type == TypeText && strings.HasPrefix(text, "ACK ") && strings.HasSuffix(text, duplicateAckSuffix), + } + if ack.Type == TypeText && strings.HasPrefix(text, "ERR ") { + return result, fmt.Errorf("%w: %s", ErrRejected, text) + } + return result, nil +} + // SendText sends a text frame. func (c *Client) SendText(text string) error { return WriteFrame(c.conn, &Frame{Type: TypeText, Payload: []byte(text)}) diff --git a/dataexchange.go b/dataexchange.go index 688b861..d6e3fe0 100644 --- a/dataexchange.go +++ b/dataexchange.go @@ -41,6 +41,12 @@ const ( // TypeFileStream INIT. Subsequent chunks are accepted only while bound to // that verified transfer ID on the same connection. TypeGovernedFileStream uint32 = 9 + // TypeTagged wraps another frame with optional correlation metadata + // (Frame.MessageID / Frame.ReplyTo). Callers never set it directly: + // WriteFrame emits it when either field is set and ReadFrame unwraps it, + // returning the inner frame with the fields populated. See tagged.go for + // the wire layout and the fallback rule for receivers that predate it. + TypeTagged uint32 = 10 ) // TraceFrame carries timing metadata around an inner message frame. @@ -129,33 +135,69 @@ type Frame struct { Type uint32 Payload []byte Filename string // only for TypeFile + + // MessageID optionally identifies this message so the receiver can drop + // an identical re-delivery and the recipient can answer it. Use + // NewMessageID to generate one. Empty means "no ID" (the pre-existing + // wire format). See tagged.go. + MessageID string + // ReplyTo optionally names the MessageID of the message this frame + // answers, so the original sender can match the reply to its request. + // Empty means "not a correlated reply". + ReplyTo string } -// WriteFrame writes a frame to a writer. +// wirePayload returns the bytes that follow the 8-byte header for f's own +// type: the payload itself, or for TypeFile the length-prefixed filename +// followed by the payload. +func wirePayload(f *Frame) ([]byte, error) { + if f.Type != TypeFile { + return f.Payload, nil + } + // Prepend filename. Validate the length BEFORE the uint16 cast: a + // name longer than 65535 bytes would wrap the 2-byte length field + // and silently truncate, and any name over maxFilenameLen is + // rejected by ReadFrame anyway — fail fast on the writer side. + name := []byte(f.Filename) + if len(name) > maxFilenameLen { + return nil, fmt.Errorf("filename too long: %d bytes (max %d)", len(name), maxFilenameLen) + } + payload := make([]byte, 2+len(name)+len(f.Payload)) + binary.BigEndian.PutUint16(payload[0:2], uint16(len(name))) + copy(payload[2:], name) + copy(payload[2+len(name):], f.Payload) + return payload, nil +} + +// WriteFrame writes a frame to a writer. A frame with MessageID or ReplyTo +// set is written as a TypeTagged wrapper around its own type; receivers that +// predate TypeTagged do not store it, so prefer Client.Send, which falls back +// to the untagged form for them. If the wrapper would push the frame over +// MaxFrameSize, WriteFrame writes nothing and returns an error wrapping +// ErrTaggedFrameTooLarge. func WriteFrame(w io.Writer, f *Frame) error { - payload := f.Payload - if f.Type == TypeFile { - // Prepend filename. Validate the length BEFORE the uint16 cast: a - // name longer than 65535 bytes would wrap the 2-byte length field - // and silently truncate, and any name over maxFilenameLen is - // rejected by ReadFrame anyway — fail fast on the writer side. - name := []byte(f.Filename) - if len(name) > maxFilenameLen { - return fmt.Errorf("filename too long: %d bytes (max %d)", len(name), maxFilenameLen) - } - payload = make([]byte, 2+len(name)+len(f.Payload)) - binary.BigEndian.PutUint16(payload[0:2], uint16(len(name))) - copy(payload[2:], name) - copy(payload[2+len(name):], f.Payload) + ftype := f.Type + var ( + payload []byte + err error + ) + if f.MessageID != "" || f.ReplyTo != "" { + payload, err = taggedWirePayload(f) + ftype = TypeTagged + } else { + payload, err = wirePayload(f) + } + if err != nil { + return err } var hdr [8]byte - binary.BigEndian.PutUint32(hdr[0:4], f.Type) + binary.BigEndian.PutUint32(hdr[0:4], ftype) binary.BigEndian.PutUint32(hdr[4:8], uint32(len(payload))) if _, err := w.Write(hdr[:]); err != nil { return err } - _, err := w.Write(payload) + _, err = w.Write(payload) return err } @@ -183,6 +225,24 @@ func ReadFrame(r io.Reader) (*Frame, error) { return nil, err } + if ftype == TypeTagged { + meta, innerType, inner, err := decodeTaggedPayload(payload) + if err != nil { + return nil, err + } + f, err := parseFramePayload(innerType, inner) + if err != nil { + return nil, err + } + f.MessageID, f.ReplyTo = meta.MessageID, meta.ReplyTo + return f, nil + } + return parseFramePayload(ftype, payload) +} + +// parseFramePayload builds a Frame from a type and the bytes that followed +// its header, splitting off and validating the TypeFile filename prefix. +func parseFramePayload(ftype uint32, payload []byte) (*Frame, error) { f := &Frame{Type: ftype, Payload: payload} if ftype == TypeFile && len(payload) >= 2 { @@ -265,6 +325,8 @@ func TypeName(t uint32) string { return "GOVERNED" case TypeGovernedFileStream: return "GOVERNED_FILESTREAM" + case TypeTagged: + return "TAGGED" default: return fmt.Sprintf("UNKNOWN(%d)", t) } diff --git a/dedupe.go b/dedupe.go new file mode 100644 index 0000000..c9d4e6f --- /dev/null +++ b/dedupe.go @@ -0,0 +1,235 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !no_dataexchange +// +build !no_dataexchange + +package dataexchange + +import ( + "context" + "crypto/sha256" + "encoding/binary" + "sync" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// Receiver-side duplicate suppression. +// +// A delivery is identified by a SHA-256 over the sender address, the frame +// type, filename, MessageID, ReplyTo and the exact payload bytes. When a +// second delivery with the same key arrives inside the window after the +// first one was stored, the service acknowledges it (with the original ACK +// plus " (duplicate)") and does not store it again. Frames with a MessageID +// are checked by default; frames without one are only checked when +// ServiceConfig.DedupeContentWindow is set, because two identical legacy +// messages from one sender can be deliberate. + +// DefaultDedupeWindow is how long a stored frame that carried a MessageID is +// remembered for duplicate suppression when ServiceConfig.DedupeWindow is 0. +const DefaultDedupeWindow = 10 * time.Minute + +// dedupeMaxEntries bounds each dedupe cache. When it is full the oldest +// remembered delivery is forgotten early, so a burst of traffic shortens the +// effective window instead of growing memory (about 100 bytes per entry). +const dedupeMaxEntries = 8192 + +type dedupeKey [sha256.Size]byte + +// deliveryDedupe remembers recently stored deliveries. Entries are created +// "in flight" when a frame starts processing, so a concurrent identical +// delivery (the same reply arriving over two paths microseconds apart) waits +// for the first instead of racing it; a failed first attempt is forgotten so +// the waiting copy is processed normally. +type deliveryDedupe struct { + window time.Duration + max int + now func() time.Time + + mu sync.Mutex + entries map[dedupeKey]*dedupeEntry + // order holds stored entries oldest first. Every entry in a cache has + // the same window, so this is also expiry order. + order []dedupeOrder + head int +} + +type dedupeEntry struct { + done chan struct{} // closed once the first delivery finished + ack string // the first delivery's ACK text (set before done closes) + expires time.Time +} + +type dedupeOrder struct { + key dedupeKey + entry *dedupeEntry +} + +func newDeliveryDedupe(window time.Duration, maxEntries int) *deliveryDedupe { + return &deliveryDedupe{ + window: window, + max: maxEntries, + now: time.Now, + entries: make(map[dedupeKey]*dedupeEntry), + } +} + +// dedupeClaim is the right (and duty) to process one delivery. The zero +// value is an inert claim. +type dedupeClaim struct { + cache *deliveryDedupe + key dedupeKey + entry *dedupeEntry +} + +// finish records the outcome of the claimed delivery: a stored delivery is +// remembered for the window (ack is replayed to duplicates), a failed one is +// forgotten so a retry is processed normally. Safe to call on a zero or +// already-finished claim. +func (c *dedupeClaim) finish(stored bool, ack string) { + if c.cache == nil { + return + } + c.cache.finish(c.key, c.entry, stored, ack) + *c = dedupeClaim{} +} + +// claim returns (claim, "", true) when the caller should process the +// delivery, or (zero, ack, true) when it duplicates a stored one, where ack +// is the first delivery's ACK text. While an identical delivery is still in +// flight it waits for that one to finish. ok is false only when ctx ends +// while waiting. +func (d *deliveryDedupe) claim(ctx context.Context, key dedupeKey) (claim dedupeClaim, duplicateOf string, ok bool) { + for { + entry, fresh := d.begin(key) + if fresh { + return dedupeClaim{cache: d, key: key, entry: entry}, "", true + } + select { + case <-entry.done: + case <-ctx.Done(): + return dedupeClaim{}, "", false + } + d.mu.Lock() + current := d.entries[key] + d.mu.Unlock() + if current == entry { + return dedupeClaim{}, entry.ack, true + } + // The first attempt failed (or its memory expired): try again. + } +} + +func (d *deliveryDedupe) begin(key dedupeKey) (*dedupeEntry, bool) { + d.mu.Lock() + defer d.mu.Unlock() + d.expireLocked(d.now()) + if entry, exists := d.entries[key]; exists { + return entry, false + } + entry := &dedupeEntry{done: make(chan struct{})} + d.entries[key] = entry + return entry, true +} + +func (d *deliveryDedupe) finish(key dedupeKey, entry *dedupeEntry, stored bool, ack string) { + d.mu.Lock() + defer d.mu.Unlock() + if stored { + entry.ack = ack + entry.expires = d.now().Add(d.window) + d.order = append(d.order, dedupeOrder{key: key, entry: entry}) + for len(d.order)-d.head > d.max { + d.popLocked() + } + } else if d.entries[key] == entry { + delete(d.entries, key) + } + close(entry.done) +} + +func (d *deliveryDedupe) expireLocked(now time.Time) { + for d.head < len(d.order) && !now.Before(d.order[d.head].entry.expires) { + d.popLocked() + } +} + +func (d *deliveryDedupe) popLocked() { + oldest := d.order[d.head] + d.order[d.head] = dedupeOrder{} + d.head++ + if d.entries[oldest.key] == oldest.entry { + delete(d.entries, oldest.key) + } + // Compact once the consumed prefix dominates, keeping the slice bounded. + if d.head > 64 && d.head*2 > len(d.order) { + d.order = append(d.order[:0], d.order[d.head:]...) + d.head = 0 + } +} + +// len reports how many deliveries (stored or in flight) are remembered. +func (d *deliveryDedupe) len() int { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.entries) +} + +// dedupeEligible reports whether frames of this (already unwrapped) type are +// stored as a single unit and therefore subject to duplicate suppression. +// Streams have their own resume protocol and trace frames are diagnostics. +func dedupeEligible(frameType uint32) bool { + switch frameType { + case TypeText, TypeJSON, TypeBinary, TypeFile, TypeGoverned: + return true + default: + return false + } +} + +// deliveryKeyFor hashes everything that makes two deliveries identical. +// Variable-length fields are length-prefixed so field boundaries cannot be +// shifted to forge a collision. +func deliveryKeyFor(from protocol.Addr, f *Frame) dedupeKey { + h := sha256.New() + var n [8]byte + writeField := func(b []byte) { + binary.BigEndian.PutUint64(n[:], uint64(len(b))) + h.Write(n[:]) + h.Write(b) + } + writeField([]byte(from.String())) + binary.BigEndian.PutUint32(n[:4], f.Type) + h.Write(n[:4]) + writeField([]byte(f.Filename)) + writeField([]byte(f.MessageID)) + writeField([]byte(f.ReplyTo)) + writeField(f.Payload) + var key dedupeKey + copy(key[:], h.Sum(nil)) + return key +} + +// dedupeCacheFor picks the cache that governs f, or nil when f is not +// subject to duplicate suppression. +func (s *Service) dedupeCacheFor(f *Frame) *deliveryDedupe { + if !dedupeEligible(f.Type) { + return nil + } + if f.MessageID != "" { + return s.dedupeByID + } + return s.dedupeByContent +} + +func (s *Service) effectiveDedupeWindow() time.Duration { + switch { + case s.cfg.DedupeWindow < 0: + return 0 + case s.cfg.DedupeWindow == 0: + return DefaultDedupeWindow + default: + return s.cfg.DedupeWindow + } +} diff --git a/inbox_budget.go b/inbox_budget.go new file mode 100644 index 0000000..f4630dc --- /dev/null +++ b/inbox_budget.go @@ -0,0 +1,296 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !no_dataexchange +// +build !no_dataexchange + +package dataexchange + +import ( + "fmt" + "log/slog" + "os" + "path/filepath" + "sort" + "sync" + "time" +) + +// Inbox capacity accounting. +// +// Two caps bound the inbox: a file count (InboxMaxFiles) and a byte total +// (InboxMaxBytes, 256 MiB by default). Oldest messages are evicted first. +// +// The byte cap is enforced before every write without listing the +// directory: the service keeps a running total, seeded by one scan and +// reconciled by the periodic eviction pass. Other processes also delete +// inbox files (`pilotctl inbox --clear`, retention sweeps), so the running +// total can only drift upward between scans; the directory is re-read before +// anything is evicted, and eviction decisions always come from that fresh +// scan, never from the running total alone. +// +// When the byte cap forces an eviction, the oldest files are removed until +// the inbox plus the incoming message fits in inboxLowWaterPercent of the +// cap, so one eviction pass makes room for many following messages instead +// of running again on every write. +// +// A write is in flight from its reservation until inboxWriteDone. Its bytes +// are counted in pending the whole time, but its file can already be on +// disk, fully or partly written, when another writer rescans. Rescans +// therefore leave in-flight files out: their bytes are not added to onDisk +// a second time, and eviction never deletes a message that has been +// admitted and is about to be acknowledged. + +// inboxLowWaterPercent is the fill level, as a percentage of the byte cap, +// that a byte-cap eviction trims the inbox down to. +const inboxLowWaterPercent = 90 + +// defaultInboxMaxFiles is the file-count cap when InboxMaxFiles is not set. +const defaultInboxMaxFiles = 10000 + +// inboxBudget is the running byte total for the inbox directory. +type inboxBudget struct { + mu sync.Mutex + dir string + // known is false until the first scan of dir. + known bool + // onDisk is the byte total of the settled inbox files (all but the + // in-flight ones) as of the last scan, plus files this service wrote + // since, minus files it removed since. + onDisk int64 + // pending is bytes admitted for writes that have not finished yet. + pending int64 + // inflight holds the destination path of every write counted in + // pending. rescanLocked skips these files. + inflight map[string]struct{} + // scans counts directory listings, so tests can check that the byte cap + // is not paid for with a full scan per message. + scans int +} + +// inboxFile is one regular file in the inbox directory. +type inboxFile struct { + name string + mod time.Time + size int64 +} + +// scanInbox lists the regular files in dir (subdirectories hold no inbox +// messages and are skipped) with their total size. +func scanInbox(dir string) ([]inboxFile, int64, error) { + entries, err := os.ReadDir(dir) + if err != nil { + return nil, 0, err + } + files := make([]inboxFile, 0, len(entries)) + var total int64 + for _, e := range entries { + if e.IsDir() { + continue + } + info, err := e.Info() + if err != nil { + continue + } + files = append(files, inboxFile{name: e.Name(), mod: info.ModTime(), size: info.Size()}) + total += info.Size() + } + return files, total, nil +} + +// inboxLowWater returns the byte level a byte-cap eviction trims to: +// inboxLowWaterPercent of maxBytes, rounded down, without overflowing. +func inboxLowWater(maxBytes int64) int64 { + return maxBytes/100*inboxLowWaterPercent + maxBytes%100*inboxLowWaterPercent/100 +} + +// removeOldestInbox deletes files from dir, oldest first, until total is at +// or below limit. files must be the result of a fresh scan. It returns the +// new total and the number of files removed; files that cannot be removed +// are skipped and still counted. +func removeOldestInbox(dir string, files []inboxFile, total, limit int64) (int64, int) { + if total <= limit { + return total, 0 + } + sort.Slice(files, func(i, j int) bool { return files[i].mod.Before(files[j].mod) }) + evicted := 0 + for i := 0; i < len(files) && total > limit; i++ { + if err := os.Remove(filepath.Join(dir, files[i].name)); err != nil { + continue + } + total -= files[i].size + evicted++ + } + return total, evicted +} + +// rescanLocked re-reads dir and resets the running total. Files of writes +// still in flight are left out of both the returned list and the total: +// their bytes are already in pending, and they must not be evicted. b.mu +// must be held. +func (b *inboxBudget) rescanLocked(dir string) ([]inboxFile, error) { + b.scans++ + files, total, err := scanInbox(dir) + if err != nil { + return nil, err + } + if len(b.inflight) > 0 { + settled := files[:0] + for _, f := range files { + if _, busy := b.inflight[filepath.Join(dir, f.name)]; busy { + total -= f.size + continue + } + settled = append(settled, f) + } + files = settled + } + b.dir, b.known, b.onDisk = dir, true, total + return files, nil +} + +// admitLocked records a reservation of size bytes for the file at path. +// b.mu must be held. +func (b *inboxBudget) admitLocked(path string, size int64) { + b.pending += size + if b.inflight == nil { + b.inflight = make(map[string]struct{}) + } + b.inflight[path] = struct{}{} +} + +// reserveInbox admits a message of size bytes, to be written to path inside +// dir, under the byte cap maxBytes (> 0), evicting the oldest messages when +// needed. On success the caller must report the write's outcome with +// inboxWriteDone(path, ...). +func (s *Service) reserveInbox(dir, path string, size, maxBytes int64) error { + b := &s.inbox + b.mu.Lock() + defer b.mu.Unlock() + if size > maxBytes { + return fmt.Errorf("inbox byte budget exceeded: a %d-byte message is larger than the %d-byte cap", size, maxBytes) + } + if size+b.pending > maxBytes { + // Even an empty inbox could not hold this message next to the writes + // already in progress; evicting would destroy messages for nothing. + return fmt.Errorf("inbox byte budget exceeded: %d bytes being written + %d > %d", b.pending, size, maxBytes) + } + if !b.known || b.dir != dir { + if _, err := b.rescanLocked(dir); err != nil { + // Best-effort, as before: an unlistable inbox does not block + // delivery; the write itself reports any real I/O problem. + slog.Warn("inbox byte budget: cannot list inbox; admitting message unchecked", "dir", dir, "err", err) + b.admitLocked(path, size) + return nil + } + } + if b.onDisk+b.pending+size > maxBytes { + // The running total may be stale (files deleted by other processes), + // so decide from a fresh listing. + files, err := b.rescanLocked(dir) + if err != nil { + slog.Warn("inbox byte budget: cannot list inbox; admitting message unchecked", "dir", dir, "err", err) + b.admitLocked(path, size) + return nil + } + if b.onDisk+b.pending+size > maxBytes { + target := inboxLowWater(maxBytes) + if size+b.pending > target { + target = maxBytes + } + before := b.onDisk + var evicted int + b.onDisk, evicted = removeOldestInbox(dir, files, b.onDisk, target-size-b.pending) + slog.Info("inbox eviction (bytes)", "dir", dir, "evicted", evicted, + "bytes_before", before, "bytes_after", b.onDisk, "max_bytes", maxBytes) + } + if b.onDisk+b.pending+size > maxBytes { + return fmt.Errorf("inbox byte budget exceeded: %d + %d > %d", b.onDisk+b.pending, size, maxBytes) + } + } + b.admitLocked(path, size) + return nil +} + +// inboxWriteDone settles the reservation reserveInbox made for path. From +// then on the file counts in onDisk (if it was written) and can be evicted. +func (s *Service) inboxWriteDone(path string, size int64, written bool) { + b := &s.inbox + b.mu.Lock() + defer b.mu.Unlock() + delete(b.inflight, path) + b.pending -= size + if b.pending < 0 { + b.pending = 0 + } + if written { + b.onDisk += size + } +} + +// inboxFileRemoved accounts for this service deleting an inbox file it wrote. +func (s *Service) inboxFileRemoved(size int64) { + b := &s.inbox + b.mu.Lock() + defer b.mu.Unlock() + b.onDisk -= size + if b.onDisk < 0 { + b.onDisk = 0 + } +} + +// evictInboxOverflow enforces both inbox caps from a fresh listing: the file +// count (InboxMaxFiles, default 10000) and the byte total +// (effectiveInboxMaxBytes, trimmed to the low-water mark when exceeded). It +// also re-seeds the running byte total. Files of writes still in flight are +// never evicted (see rescanLocked). Best-effort: I/O errors are logged and +// skipped. Called periodically from saveInboxMessage. +func (s *Service) evictInboxOverflow(dir string) { + b := &s.inbox + b.mu.Lock() + defer b.mu.Unlock() + files, err := b.rescanLocked(dir) + if err != nil { + slog.Debug("inbox evict: readdir", "dir", dir, "err", err) + return + } + maxFiles := s.cfg.InboxMaxFiles + if maxFiles <= 0 { + maxFiles = defaultInboxMaxFiles + } + if len(files) > maxFiles { + // Oldest first. + sort.Slice(files, func(i, j int) bool { return files[i].mod.Before(files[j].mod) }) + toEvict := len(files) - maxFiles + evicted := 0 + for i := 0; i < toEvict; i++ { + if err := os.Remove(filepath.Join(dir, files[i].name)); err != nil { + continue + } + b.onDisk -= files[i].size + evicted++ + } + files = files[toEvict:] + slog.Info("inbox eviction", "dir", dir, "evicted", evicted, "remaining", len(files)) + } + if maxBytes := s.effectiveInboxMaxBytes(); maxBytes > 0 && b.onDisk+b.pending > maxBytes { + // Writes still in flight were admitted under the cap and are not + // candidates, so trim the settled files until they fit next to them: + // down to the low-water mark, or to the cap when the in-flight + // writes alone are above the low-water mark (as in reserveInbox). + target := inboxLowWater(maxBytes) + if b.pending > target { + target = maxBytes + } + before := b.onDisk + var evicted int + b.onDisk, evicted = removeOldestInbox(dir, files, b.onDisk, target-b.pending) + slog.Info("inbox eviction (bytes)", "dir", dir, "evicted", evicted, + "bytes_before", before, "bytes_after", b.onDisk, "max_bytes", maxBytes) + } +} + +// inboxTotalBytes sums the on-disk size of all regular files in dir. +func inboxTotalBytes(dir string) (int64, error) { + _, total, err := scanInbox(dir) + return total, err +} diff --git a/service.go b/service.go index d772336..84cfd0a 100644 --- a/service.go +++ b/service.go @@ -14,7 +14,6 @@ import ( "log/slog" "os" "path/filepath" - "sort" "sync/atomic" "time" "unicode/utf8" @@ -40,11 +39,27 @@ type ServiceConfig struct { // misbehaving peer or sustained inbound load fills the operator's // disk indefinitely. InboxMaxFiles int - // InboxMaxBytes caps the total on-disk bytes used by the inbox. - // saveInboxMessage checks the accumulated size before every write and - // evicts oldest-first when over. Zero ⇒ DefaultInboxMaxBytes; a - // negative value disables the byte cap entirely (escape hatch). + // InboxMaxBytes caps the total on-disk bytes used by the inbox. Every + // write is checked against a running total; when a message would not + // fit, the oldest messages are evicted until the inbox plus the new + // message is at or below 90% of the cap. A single message larger than the + // cap is rejected without evicting anything. Zero ⇒ DefaultInboxMaxBytes; + // a negative value disables the byte cap entirely (escape hatch). InboxMaxBytes int64 + // DedupeWindow is how long a stored frame that carried a MessageID is + // remembered, so an identical re-delivery (same sender, MessageID, + // ReplyTo, type, filename and bytes) is acknowledged without being + // stored again. Zero ⇒ DefaultDedupeWindow (10 min); negative disables. + // Frames without a MessageID are unaffected unless DedupeContentWindow + // is set. + DedupeWindow time.Duration + // DedupeContentWindow, when positive, also suppresses byte-identical + // frames WITHOUT a MessageID from the same sender that arrive within this + // window (for example a peer that answers one request over two paths). + // Zero (the default) keeps the historical behaviour of storing every + // such frame: two identical legacy messages can be deliberate, and a + // sender cannot mark them otherwise. Keep it short (a few seconds). + DedupeContentWindow time.Duration // ReceivedMaxBytes caps the total on-disk bytes used by the received- // files directory (completed TypeFile / TypeFileStream files plus any // retained .partial fragments). Enforced before every write so a peer @@ -143,8 +158,10 @@ func (s *Service) effectiveIdleTimeout() time.Duration { // inboxEvictCheckEvery: only run the eviction-scan once every N saves // — full readdir + sort is O(n log n), so we don't want it on every -// write. The cap is a soft cap; transient overshoot of up to this -// many files between checks is acceptable. +// write. The file-count cap is a soft cap; transient overshoot of up to +// this many files between checks is acceptable. The byte cap is enforced on +// every write from a running total (see inbox_budget.go); this pass also +// re-seeds that total from disk. const inboxEvictCheckEvery = 64 // Service is the L11 plugin adapter. Daemon (L7) holds it only as @@ -158,6 +175,15 @@ type Service struct { seq atomic.Uint64 retention *governedRetentionManager replay *governedReplayGuard + inbox inboxBudget + // dedupeByID remembers stored frames that carried a MessageID; + // dedupeByContent (nil unless DedupeContentWindow > 0) remembers frames + // without one. See dedupe.go. + dedupeByID *deliveryDedupe + dedupeByContent *deliveryDedupe + // afterInboxWrite, when set (tests only), runs right after an inbox + // message's file is written, while its byte reservation is still open. + afterInboxWrite func(path string) } // persistedDelivery separates a disk write from its visible notification so a @@ -170,7 +196,14 @@ type persistedDelivery struct { } func NewService(cfg ServiceConfig) *Service { - return &Service{cfg: cfg, replay: newGovernedReplayGuard()} + s := &Service{cfg: cfg, replay: newGovernedReplayGuard()} + if window := s.effectiveDedupeWindow(); window > 0 { + s.dedupeByID = newDeliveryDedupe(window, dedupeMaxEntries) + } + if cfg.DedupeContentWindow > 0 { + s.dedupeByContent = newDeliveryDedupe(cfg.DedupeContentWindow, dedupeMaxEntries) + } + return s } func (s *Service) Name() string { return "dataexchange" } @@ -321,7 +354,12 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { var sr *StreamReceiver governedStreams := make(map[[transferIDLen]byte]GovernedStreamInit) streamRetention := make(map[[transferIDLen]byte]retentionTicket) + // claim is the dedupe claim for the frame being processed. If handling + // unwinds early (a panic), releasing it as "not stored" lets any + // identical delivery waiting on it proceed instead of hanging. + var claim dedupeClaim defer func() { + claim.finish(false, "") for _, ticket := range streamRetention { _ = ticket.rollback() } @@ -409,10 +447,37 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { if err != nil { return } - slog.Debug("dataexchange frame received", + slog.Debug("dataexchange frame received", append([]any{ "type", TypeName(frame.Type), "bytes", len(frame.Payload), - "remote", conn.RemoteAddr()) + "remote", conn.RemoteAddr()}, correlationAttrs(frame)...)...) + + // Correlation metadata rides outside any governed envelope, so keep + // it to re-attach after the envelope is unwrapped below. + messageID, replyTo := frame.MessageID, frame.ReplyTo + + // Duplicate suppression: an identical delivery of a frame that was + // already stored is acknowledged again but not stored twice. + if cache := s.dedupeCacheFor(frame); cache != nil { + var ( + duplicateAck string + ok bool + ) + claim, duplicateAck, ok = cache.claim(ctx, deliveryKeyFor(conn.RemoteAddr(), frame)) + if !ok { + return + } + if claim.cache == nil { + slog.Debug("dataexchange duplicate delivery suppressed", append([]any{ + "type", TypeName(frame.Type), + "bytes", len(frame.Payload), + "remote", conn.RemoteAddr()}, correlationAttrs(frame)...)...) + if werr := WriteFrame(conn, &Frame{Type: TypeText, Payload: []byte(duplicateAck + duplicateAckSuffix)}); werr != nil { + return + } + continue + } + } var ( saveErr error @@ -432,6 +497,7 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { } else { governed = &decoded frame = governed.DataFrame() + frame.MessageID, frame.ReplyTo = messageID, replyTo } } else if frame.Type == TypeGovernedFileStream { decoded, governErr := DecodeGovernedStreamInit(frame) @@ -547,7 +613,7 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { Payload: []byte(fmt.Sprintf("ERR trace parse: %v", tferr)), } } else { - innerFrame := &Frame{Type: tf.InnerType, Payload: tf.Payload} + innerFrame := &Frame{Type: tf.InnerType, Payload: tf.Payload, MessageID: messageID, ReplyTo: replyTo} innerSaveErr := s.saveInboxMessage(innerFrame, conn.RemoteAddr()) inboxWrittenAtNs := time.Now().UnixNano() innerAck := fmt.Sprintf("ACK %s %d bytes", TypeName(tf.InnerType), len(tf.Payload)) @@ -593,6 +659,7 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { } ackFrame = &Frame{Type: TypeText, Payload: []byte(ackMsg)} } + claim.finish(saveErr == nil, string(ackFrame.Payload)) if err := WriteFrame(conn, ackFrame); err != nil { if s.deps.Events != nil { s.deps.Events.Publish("dataexchange.ack_failed", map[string]any{ @@ -769,11 +836,13 @@ func (s *Service) prepareReceivedFile(frame *Frame, disclosure *decision.Disclos return retentionErr }, commit: func() { - slog.Info("file saved", "path", destPath, "bytes", len(frame.Payload)) + slog.Info("file saved", append([]any{"path", destPath, "bytes", len(frame.Payload)}, correlationAttrs(frame)...)...) if s.deps.Events != nil { - s.deps.Events.Publish("file.received", map[string]any{ + event := map[string]any{ "filename": safeName, "size": len(frame.Payload), "path": destPath, - }) + } + addCorrelation(event, frame) + s.deps.Events.Publish("file.received", event) } }, }, nil @@ -797,36 +866,6 @@ func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosu return persistedDelivery{}, fmt.Errorf("mkdir: %w", err) } - // Byte-budget check: confirm there is room BEFORE writing. Evict if - // over, then re-check. The cap is always on (defaulted) unless the - // operator explicitly disables it with a negative config value. - if maxBytes := s.effectiveInboxMaxBytes(); maxBytes > 0 { - current, _ := inboxTotalBytes(dir) - // Estimate the JSON overhead: type, from, data, bytes, received_at, - // and possibly data_b64. - estimated := int64(len(frame.Payload)) + 256 - if current+estimated > maxBytes { - s.evictInboxOverflowByBytes(dir) - after, _ := inboxTotalBytes(dir) - if after+estimated > maxBytes { - slog.Warn("inbox byte budget exceeded after eviction", - "current_bytes", after, - "max_bytes", maxBytes, - "frame_bytes", len(frame.Payload)) - if s.deps.Events != nil { - s.deps.Events.Publish("inbox.full", map[string]any{ - "from": from.String(), - "type": TypeName(frame.Type), - "frame_bytes": len(frame.Payload), - "max_bytes": maxBytes, - }) - } - return persistedDelivery{}, fmt.Errorf("inbox byte budget exceeded: %d + %d > %d", - after, estimated, maxBytes) - } - } - } - ts := time.Now() msg := map[string]interface{}{ "type": TypeName(frame.Type), @@ -834,6 +873,9 @@ func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosu "bytes": len(frame.Payload), "received_at": ts.Format(time.RFC3339Nano), } + // Correlation metadata, only when the sender supplied it: a service + // answers a message by echoing its message_id as reply_to. + addCorrelation(msg, frame) // Store the payload losslessly. JSON cannot represent arbitrary bytes in // a string — invalid UTF-8 is replaced with U+FFFD by encoding/json — so // any payload that is not valid UTF-8 (all binary frames, in practice) @@ -852,10 +894,42 @@ func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosu if err != nil { return persistedDelivery{}, fmt.Errorf("marshal: %w", err) } + size := int64(len(data)) seq := s.seq.Add(1) filename := fmt.Sprintf("%s-%s-%06d.json", TypeName(frame.Type), ts.Format("20060102-150405.000"), seq) destPath := filepath.Join(dir, filename) + + // Byte budget: admit the exact file size BEFORE writing, evicting the + // oldest messages if needed. The cap is always on (defaulted) unless the + // operator explicitly disables it with a negative config value. Until + // inboxWriteDone settles it, the reservation also keeps destPath from + // being evicted by a concurrent writer. + maxBytes := s.effectiveInboxMaxBytes() + if maxBytes > 0 { + if err := s.reserveInbox(dir, destPath, size, maxBytes); err != nil { + slog.Warn("inbox byte budget exceeded", + "max_bytes", maxBytes, + "message_bytes", size, + "err", err) + if s.deps.Events != nil { + s.deps.Events.Publish("inbox.full", map[string]any{ + "from": from.String(), + "type": TypeName(frame.Type), + "frame_bytes": len(frame.Payload), + "max_bytes": maxBytes, + }) + } + return persistedDelivery{}, err + } + } + written := false + defer func() { + if maxBytes > 0 { + s.inboxWriteDone(destPath, size, written) + } + }() + retention, err := s.prepareGovernedRetention(disclosure, destPath) if err != nil { return persistedDelivery{}, err @@ -864,9 +938,16 @@ func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosu _ = retention.rollback() return persistedDelivery{}, fmt.Errorf("write: %w", err) } + written = true + if s.afterInboxWrite != nil { + s.afterInboxWrite(destPath) + } return persistedDelivery{ rollback: func() error { removeErr := os.Remove(destPath) + if removeErr == nil && maxBytes > 0 { + s.inboxFileRemoved(size) + } retentionErr := retention.rollback() if removeErr != nil && !os.IsNotExist(removeErr) { return removeErr @@ -874,12 +955,14 @@ func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosu return retentionErr }, commit: func() { - slog.Info("inbox message saved", "path", destPath, "type", TypeName(frame.Type), "bytes", len(frame.Payload)) + slog.Info("inbox message saved", append([]any{"path", destPath, "type", TypeName(frame.Type), "bytes", len(frame.Payload)}, correlationAttrs(frame)...)...) if s.deps.Events != nil { - s.deps.Events.Publish("message.received", map[string]any{ + event := map[string]any{ "type": TypeName(frame.Type), "from": from.String(), "size": len(frame.Payload), - }) + } + addCorrelation(event, frame) + s.deps.Events.Publish("message.received", event) } // Periodic eviction so a misbehaving peer (or sustained inbound // load) cannot fill the operator's disk. We sample every @@ -891,115 +974,29 @@ func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosu }, nil } -// evictInboxOverflow trims the inbox to at most cfg.InboxMaxFiles by -// deleting the oldest files (by mtime). When InboxMaxBytes > 0, the -// eviction target is total bytes rather than file count. Best-effort: -// I/O errors are logged and the loop continues. Called periodically -// from saveInboxMessage. -func (s *Service) evictInboxOverflow(dir string) { - // Byte-based eviction when InboxMaxBytes is configured. - if s.cfg.InboxMaxBytes > 0 { - s.evictInboxOverflowByBytes(dir) - return - } - maxFiles := s.cfg.InboxMaxFiles - if maxFiles <= 0 { - maxFiles = 10000 - } - entries, err := os.ReadDir(dir) - if err != nil { - slog.Debug("inbox evict: readdir", "dir", dir, "err", err) - return - } - // Filter out subdirectories — they consume no inbox capacity. - type aged struct { - name string - mod time.Time - } - files := make([]aged, 0, len(entries)) - for _, e := range entries { - if e.IsDir() { - continue - } - info, err := e.Info() - if err != nil { - continue - } - files = append(files, aged{name: e.Name(), mod: info.ModTime()}) +// correlationAttrs returns slog attributes for a frame's MessageID / ReplyTo, +// omitting absent ones so untagged traffic logs exactly as before. +func correlationAttrs(frame *Frame) []any { + var attrs []any + if frame.MessageID != "" { + attrs = append(attrs, "message_id", frame.MessageID) } - if len(files) <= maxFiles { - return + if frame.ReplyTo != "" { + attrs = append(attrs, "reply_to", frame.ReplyTo) } - // Oldest first. - sort.Slice(files, func(i, j int) bool { return files[i].mod.Before(files[j].mod) }) - toEvict := len(files) - maxFiles - for i := 0; i < toEvict; i++ { - _ = os.Remove(filepath.Join(dir, files[i].name)) - } - slog.Info("inbox eviction", "dir", dir, "evicted", toEvict, "remaining", maxFiles) + return attrs } -// evictInboxOverflowByBytes trims total inbox size to InboxMaxBytes. -func (s *Service) evictInboxOverflowByBytes(dir string) { - entries, err := os.ReadDir(dir) - if err != nil { - slog.Debug("inbox evict: readdir", "dir", dir, "err", err) - return +// addCorrelation copies a frame's optional MessageID / ReplyTo into an inbox +// record or event payload under the "message_id" / "reply_to" keys. Absent +// values are omitted, so untagged traffic produces exactly the old fields. +func addCorrelation(m map[string]any, frame *Frame) { + if frame.MessageID != "" { + m["message_id"] = frame.MessageID } - type aged struct { - name string - mod time.Time - size int64 - } - files := make([]aged, 0, len(entries)) - var total int64 - for _, e := range entries { - if e.IsDir() { - continue - } - info, err := e.Info() - if err != nil { - continue - } - sz := info.Size() - files = append(files, aged{name: e.Name(), mod: info.ModTime(), size: sz}) - total += sz - } - if total <= s.cfg.InboxMaxBytes { - return - } - // Oldest first. - sort.Slice(files, func(i, j int) bool { return files[i].mod.Before(files[j].mod) }) - evicted := 0 - for i := 0; i < len(files) && total > s.cfg.InboxMaxBytes; i++ { - p := filepath.Join(dir, files[i].name) - if err := os.Remove(p); err != nil { - continue - } - total -= files[i].size - evicted++ - } - slog.Info("inbox eviction (bytes)", "dir", dir, "evicted", evicted, "total_bytes_after", total, "max_bytes", s.cfg.InboxMaxBytes) -} - -// inboxTotalBytes sums the on-disk size of all regular files in dir. -func inboxTotalBytes(dir string) (int64, error) { - entries, err := os.ReadDir(dir) - if err != nil { - return 0, err - } - var total int64 - for _, e := range entries { - if e.IsDir() { - continue - } - info, err := e.Info() - if err != nil { - continue - } - total += info.Size() + if frame.ReplyTo != "" { + m["reply_to"] = frame.ReplyTo } - return total, nil } // dirTotalBytes sums the on-disk size of every regular file under dir, diff --git a/service_disabled.go b/service_disabled.go index 066004c..19b37d1 100644 --- a/service_disabled.go +++ b/service_disabled.go @@ -28,6 +28,8 @@ type ServiceConfig struct { IncludeBase64 bool InboxMaxFiles int InboxMaxBytes int64 + DedupeWindow time.Duration + DedupeContentWindow time.Duration ReceivedMaxBytes int64 IdleTimeout time.Duration RequireGoverned bool diff --git a/tagged.go b/tagged.go new file mode 100644 index 0000000..1533c2f --- /dev/null +++ b/tagged.go @@ -0,0 +1,175 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package dataexchange + +import ( + "crypto/rand" + "encoding/binary" + "encoding/hex" + "encoding/json" + "errors" + "fmt" +) + +// Request/reply correlation on the wire. +// +// A frame that carries a MessageID or ReplyTo travels as a TypeTagged frame: +// +// [4-byte TypeTagged][4-byte length][payload] +// payload = [2-byte header_len][header: UTF-8 JSON object] +// [4-byte inner_type][inner payload] +// +// The header is a JSON object such as +// +// {"message_id":"4f1c…","reply_to":"9b2e…"} +// +// Both keys are optional; unknown keys are ignored so later versions can add +// fields. The inner payload is exactly what the inner type carries on its +// own (a TypeFile inner keeps its [2-byte name length][name] prefix), and +// the inner type may be any type except TypeTagged itself. +// +// Compatibility: a frame without a MessageID or ReplyTo is written in the +// original format, byte for byte, so nothing changes for peers that do not +// use correlation. A receiver that predates TypeTagged stores nothing for a +// tagged frame and answers "ACK UNKNOWN(10) ..." (v0.2.1 and older, which +// silently drop unknown types) or "ERR UNKNOWN(10) ..." (v0.2.2 and later); +// Client.Send recognises both and re-sends the same frame untagged on the +// same connection. +// +// Size: the tagged wrapper adds 6 bytes plus the JSON header (at most +// MaxTaggedOverhead bytes in all), and the whole wrapped frame must fit in +// MaxFrameSize, because the receiver checks the outer length. WriteFrame +// returns ErrTaggedFrameTooLarge rather than write a tagged frame that does +// not fit; Client.Send then sends the frame untagged. + +// MaxMessageIDLen is the longest MessageID or ReplyTo value accepted on the +// wire. IDs are 1..MaxMessageIDLen bytes of ASCII letters, digits, '-', '_', +// '.' or ':' — enough for UUIDs, ULIDs and hex nonces, and safe to embed in +// file names, JSON and log lines without escaping. +const MaxMessageIDLen = 128 + +// maxTaggedHeaderLen bounds the JSON header of a TypeTagged frame. Two +// maximal IDs plus their keys fit comfortably; the cap keeps a hostile peer +// from making the receiver parse a large JSON document per frame. +const maxTaggedHeaderLen = 1024 + +// MaxTaggedOverhead is the most a MessageID/ReplyTo header adds to a frame's +// payload: the 2-byte header length, the JSON header with two +// MaxMessageIDLen IDs ({"message_id":"…","reply_to":"…"}, 287 bytes) and the +// 4-byte inner type. A frame stays taggable if its untagged payload is at +// most MaxFrameSize - MaxTaggedOverhead bytes. +const MaxTaggedOverhead = 2 + len(`{"message_id":"","reply_to":""}`) + 2*MaxMessageIDLen + 4 + +// ErrTaggedFrameTooLarge is returned (wrapped) by WriteFrame when a frame +// with a MessageID or ReplyTo would exceed MaxFrameSize once its header is +// added. Nothing is written in that case. Client.Send handles it by sending +// the frame untagged, which succeeds when the payload alone fits; a raw +// WriteFrame caller can do the same, or shrink the payload by up to +// MaxTaggedOverhead bytes. +var ErrTaggedFrameTooLarge = errors.New("dataexchange: frame with MessageID/ReplyTo exceeds MaxFrameSize") + +// taggedHeader is the JSON header of a TypeTagged frame. +type taggedHeader struct { + MessageID string `json:"message_id,omitempty"` + ReplyTo string `json:"reply_to,omitempty"` +} + +// NewMessageID returns a random 128-bit message ID as 32 lowercase hex +// characters, suitable for Frame.MessageID. +func NewMessageID() string { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + // crypto/rand.Read does not fail on supported platforms (Go 1.24+ + // crashes the program instead of returning an error). + panic(fmt.Sprintf("dataexchange: crypto/rand: %v", err)) + } + return hex.EncodeToString(b[:]) +} + +// ValidMessageID reports whether id is acceptable as a Frame.MessageID or +// Frame.ReplyTo value on the wire. +func ValidMessageID(id string) bool { + if len(id) == 0 || len(id) > MaxMessageIDLen { + return false + } + for i := 0; i < len(id); i++ { + c := id[i] + switch { + case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9': + case c == '-', c == '_', c == '.', c == ':': + default: + return false + } + } + return true +} + +func validateTaggedHeader(h taggedHeader) error { + if h.MessageID != "" && !ValidMessageID(h.MessageID) { + return fmt.Errorf("invalid message_id: want 1-%d bytes of [A-Za-z0-9._:-]", MaxMessageIDLen) + } + if h.ReplyTo != "" && !ValidMessageID(h.ReplyTo) { + return fmt.Errorf("invalid reply_to: want 1-%d bytes of [A-Za-z0-9._:-]", MaxMessageIDLen) + } + return nil +} + +// taggedWirePayload encodes f (which carries a MessageID and/or ReplyTo) as +// the payload of a TypeTagged frame. +func taggedWirePayload(f *Frame) ([]byte, error) { + if f.Type == TypeTagged { + return nil, fmt.Errorf("dataexchange: a TypeTagged frame cannot itself carry MessageID/ReplyTo") + } + h := taggedHeader{MessageID: f.MessageID, ReplyTo: f.ReplyTo} + if err := validateTaggedHeader(h); err != nil { + return nil, fmt.Errorf("dataexchange: %w", err) + } + header, err := json.Marshal(h) + if err != nil { + return nil, fmt.Errorf("dataexchange: marshal tagged header: %w", err) + } + inner, err := wirePayload(f) + if err != nil { + return nil, err + } + // Check before allocating: the receiver rejects an outer length over + // MaxFrameSize by dropping the connection, which the sender could not + // tell apart from a network failure. + size := uint64(2+len(header)+4) + uint64(len(inner)) + if size > uint64(MaxFrameSize) { + return nil, fmt.Errorf("%w: %d bytes tagged (%d untagged), max %d", ErrTaggedFrameTooLarge, size, len(inner), MaxFrameSize) + } + out := make([]byte, size) + binary.BigEndian.PutUint16(out[0:2], uint16(len(header))) + copy(out[2:], header) + binary.BigEndian.PutUint32(out[2+len(header):], f.Type) + copy(out[2+len(header)+4:], inner) + return out, nil +} + +// decodeTaggedPayload splits a TypeTagged payload into its header, inner +// type and inner payload. The inner payload aliases the input. +func decodeTaggedPayload(payload []byte) (taggedHeader, uint32, []byte, error) { + var h taggedHeader + if len(payload) < 2 { + return h, 0, nil, fmt.Errorf("tagged frame too short: %d bytes", len(payload)) + } + headerLen := int(binary.BigEndian.Uint16(payload[0:2])) + if headerLen == 0 || headerLen > maxTaggedHeaderLen { + return h, 0, nil, fmt.Errorf("tagged frame header length %d out of range (1-%d)", headerLen, maxTaggedHeaderLen) + } + if len(payload) < 2+headerLen+4 { + return h, 0, nil, fmt.Errorf("tagged frame truncated: %d bytes for a %d-byte header", len(payload), headerLen) + } + if err := json.Unmarshal(payload[2:2+headerLen], &h); err != nil { + return taggedHeader{}, 0, nil, fmt.Errorf("tagged frame header: %w", err) + } + if err := validateTaggedHeader(h); err != nil { + return taggedHeader{}, 0, nil, fmt.Errorf("tagged frame header: %w", err) + } + innerType := binary.BigEndian.Uint32(payload[2+headerLen : 2+headerLen+4]) + if innerType == TypeTagged { + return taggedHeader{}, 0, nil, fmt.Errorf("tagged frame cannot wrap another tagged frame") + } + return h, innerType, payload[2+headerLen+4:], nil +} diff --git a/zz_correlation_test.go b/zz_correlation_test.go new file mode 100644 index 0000000..5a363e3 --- /dev/null +++ b/zz_correlation_test.go @@ -0,0 +1,877 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !no_dataexchange +// +build !no_dataexchange + +package dataexchange + +import ( + "context" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/pilot-protocol/common/coreapi" + "github.com/pilot-protocol/common/decision" + "github.com/pilot-protocol/common/protocol" +) + +// addrPipeStream is a pipeStream whose remote (sender) address is chosen by +// the test, so several connections can come from the same or different +// peers. +type addrPipeStream struct { + *pipeStream + remote protocol.Addr +} + +func (s *addrPipeStream) RemoteAddr() coreapi.Addr { return s.remote } + +// clientEnd is the sender's side of a service connection. +type clientEnd struct { + r *io.PipeReader + w *io.PipeWriter +} + +func (c clientEnd) Read(p []byte) (int, error) { return c.r.Read(p) } +func (c clientEnd) Write(p []byte) (int, error) { return c.w.Write(p) } + +// openServiceConn runs svc.handleConn for one connection from remote and +// returns the sender's end. The connection is torn down at test cleanup. +func openServiceConn(t *testing.T, svc *Service, remote protocol.Addr) clientEnd { + t.Helper() + c2sR, c2sW := io.Pipe() + s2cR, s2cW := io.Pipe() + stream := &addrPipeStream{pipeStream: newPipeStream(c2sR, s2cW), remote: remote} + done := make(chan struct{}) + go func() { + svc.handleConn(context.Background(), stream) + close(done) + }() + t.Cleanup(func() { + _ = c2sW.Close() + <-done + _ = s2cR.Close() + }) + return clientEnd{r: s2cR, w: c2sW} +} + +// inboxRecords returns the inbox JSON records in dir, oldest file first. +func inboxRecords(t *testing.T, dir string) []map[string]any { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("readdir: %v", err) + } + names := make([]string, 0, len(entries)) + for _, e := range entries { + if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { + names = append(names, e.Name()) + } + } + // Names are --.json; order by the sequence number. + seqOf := func(name string) string { return name[strings.LastIndex(name, "-")+1:] } + sort.Slice(names, func(i, j int) bool { return seqOf(names[i]) < seqOf(names[j]) }) + out := make([]map[string]any, 0, len(names)) + for _, name := range names { + raw, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + t.Fatalf("read %s: %v", name, err) + } + var rec map[string]any + if err := json.Unmarshal(raw, &rec); err != nil { + t.Fatalf("decode %s: %v", name, err) + } + out = append(out, rec) + } + return out +} + +var ( + peerA = protocol.Addr{Network: 1, Node: 0xA} + peerB = protocol.Addr{Network: 1, Node: 0xB} +) + +func mustSend(t *testing.T, conn io.ReadWriter, f *Frame) *SendResult { + t.Helper() + res, err := sendAndAwaitAck(conn, f) + if err != nil { + t.Fatalf("send %+v: %v", f, err) + } + return res +} + +func TestService_TaggedMessageRecordsCorrelation(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + events := newCapturingEvents() + svc := NewService(ServiceConfig{InboxDir: tmp}) + svc.deps = coreapi.Deps{Events: events} + conn := openServiceConn(t, svc, peerA) + + res := mustSend(t, conn, &Frame{Type: TypeText, Payload: []byte("weather?"), MessageID: "req-1"}) + if !res.Tagged || res.Duplicate || string(res.Ack.Payload) != "ACK TEXT 8 bytes" { + t.Fatalf("request result = %+v ack=%q", res, res.Ack.Payload) + } + res = mustSend(t, conn, &Frame{Type: TypeJSON, Payload: []byte(`{"t":21}`), ReplyTo: "req-1"}) + if !res.Tagged || string(res.Ack.Payload) != "ACK JSON 8 bytes" { + t.Fatalf("reply result = %+v ack=%q", res, res.Ack.Payload) + } + + recs := inboxRecords(t, tmp) + if len(recs) != 2 { + t.Fatalf("inbox records = %d, want 2", len(recs)) + } + if recs[0]["message_id"] != "req-1" || recs[0]["data"] != "weather?" { + t.Fatalf("request record = %v", recs[0]) + } + if _, has := recs[0]["reply_to"]; has { + t.Fatalf("request record has a reply_to: %v", recs[0]) + } + if recs[1]["reply_to"] != "req-1" || recs[1]["type"] != "JSON" { + t.Fatalf("reply record = %v", recs[1]) + } + if _, has := recs[1]["message_id"]; has { + t.Fatalf("reply record has a message_id: %v", recs[1]) + } + + events.mu.Lock() + defer events.mu.Unlock() + var got []map[string]any + for _, e := range events.published { + if e.topic == "message.received" { + got = append(got, e.payload) + } + } + if len(got) != 2 || got[0]["message_id"] != "req-1" || got[1]["reply_to"] != "req-1" { + t.Fatalf("message.received events = %v", got) + } +} + +// TestService_UntaggedRecordUnchanged: an old sender's frame produces exactly +// the fields it always did. +func TestService_UntaggedRecordUnchanged(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + conn := openServiceConn(t, NewService(ServiceConfig{InboxDir: tmp}), peerA) + if _, err := conn.Write(rawFrame(TypeText, []byte("legacy"))); err != nil { + t.Fatal(err) + } + ack, err := ReadFrame(conn) + if err != nil || string(ack.Payload) != "ACK TEXT 6 bytes" { + t.Fatalf("ack=%v err=%v", ack, err) + } + recs := inboxRecords(t, tmp) + if len(recs) != 1 { + t.Fatalf("records = %d", len(recs)) + } + keys := make([]string, 0, len(recs[0])) + for k := range recs[0] { + keys = append(keys, k) + } + sort.Strings(keys) + if got := strings.Join(keys, ","); got != "bytes,data,data_encoding,from,received_at,type" { + t.Fatalf("untagged record keys = %s", got) + } +} + +func TestService_DuplicateTaggedDeliveryStoredOnce(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + svc := NewService(ServiceConfig{InboxDir: tmp}) + frame := &Frame{Type: TypeText, Payload: []byte("same bytes"), MessageID: "m-42"} + + first := mustSend(t, openServiceConn(t, svc, peerA), frame) + if first.Duplicate { + t.Fatal("first delivery reported as duplicate") + } + // Re-sends on one connection (a retry), and on another connection (a + // relay path next to a direct one), from the same peer. + retryConn := openServiceConn(t, svc, peerA) + for i, conn := range []io.ReadWriter{retryConn, retryConn, openServiceConn(t, svc, peerA)} { + res := mustSend(t, conn, frame) + if !res.Duplicate || string(res.Ack.Payload) != "ACK TEXT 10 bytes (duplicate)" { + t.Fatalf("re-delivery %d: result=%+v ack=%q", i, res, res.Ack.Payload) + } + } + if n := len(inboxRecords(t, tmp)); n != 1 { + t.Fatalf("inbox records = %d, want 1", n) + } + + // The same bytes and ID from a different peer are a different message. + if res := mustSend(t, openServiceConn(t, svc, peerB), frame); res.Duplicate { + t.Fatal("another peer's message was suppressed") + } + // Same ID, different payload: not an identical delivery. + if res := mustSend(t, retryConn, &Frame{Type: TypeText, Payload: []byte("other bytes"), MessageID: "m-42"}); res.Duplicate { + t.Fatal("a different payload under the same ID was suppressed") + } + // Same payload, different type: not identical either. + if res := mustSend(t, retryConn, &Frame{Type: TypeJSON, Payload: []byte("same bytes"), MessageID: "m-42"}); res.Duplicate { + t.Fatal("a different frame type was suppressed") + } + if n := len(inboxRecords(t, tmp)); n != 4 { + t.Fatalf("inbox records = %d, want 4", n) + } +} + +// TestService_UntaggedDuplicatesStoredByDefault: without a MessageID and +// without DedupeContentWindow, identical frames are all stored (old +// behaviour). +func TestService_UntaggedDuplicatesStoredByDefault(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + conn := openServiceConn(t, NewService(ServiceConfig{InboxDir: tmp}), peerA) + for i := 0; i < 3; i++ { + if res := mustSend(t, conn, &Frame{Type: TypeText, Payload: []byte("ok")}); res.Duplicate { + t.Fatalf("send %d suppressed", i) + } + } + if n := len(inboxRecords(t, tmp)); n != 3 { + t.Fatalf("inbox records = %d, want 3", n) + } +} + +func TestService_ContentDedupeWindow(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + svc := NewService(ServiceConfig{InboxDir: tmp, DedupeContentWindow: 2 * time.Second}) + now := time.Unix(1_700_000_000, 0) + var clockMu sync.Mutex + svc.dedupeByContent.now = func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return now } + advance := func(d time.Duration) { clockMu.Lock(); now = now.Add(d); clockMu.Unlock() } + conn := openServiceConn(t, svc, peerA) + + // The list-agents pattern: one reply delivered twice within a second. + reply := &Frame{Type: TypeJSON, Payload: []byte(`{"items":[1]}`)} + mustSend(t, conn, reply) + advance(700 * time.Millisecond) + if res := mustSend(t, conn, reply); !res.Duplicate { + t.Fatal("identical untagged frame inside the window was stored again") + } + // A ReplyTo-only frame is content-deduplicated too, keyed with its ReplyTo. + correlated := &Frame{Type: TypeJSON, Payload: []byte(`{"items":[1]}`), ReplyTo: "req-9"} + if res := mustSend(t, conn, correlated); res.Duplicate { + t.Fatal("a reply to a different request was suppressed") + } + if res := mustSend(t, conn, correlated); !res.Duplicate { + t.Fatal("identical ReplyTo-only frame inside the window was stored again") + } + // After the window, the same bytes are a new message. + advance(3 * time.Second) + if res := mustSend(t, conn, reply); res.Duplicate { + t.Fatal("identical frame after the window was suppressed") + } + if n := len(inboxRecords(t, tmp)); n != 3 { + t.Fatalf("inbox records = %d, want 3", n) + } +} + +func TestService_DedupeWindowNegativeDisables(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + svc := NewService(ServiceConfig{InboxDir: tmp, DedupeWindow: -1}) + if svc.dedupeByID != nil || svc.dedupeByContent != nil { + t.Fatal("dedupe caches created while disabled") + } + conn := openServiceConn(t, svc, peerA) + frame := &Frame{Type: TypeText, Payload: []byte("x"), MessageID: "m"} + mustSend(t, conn, frame) + if res := mustSend(t, conn, frame); res.Duplicate { + t.Fatal("dedupe ran although DedupeWindow < 0") + } + if n := len(inboxRecords(t, tmp)); n != 2 { + t.Fatalf("inbox records = %d, want 2", n) + } +} + +// TestService_ConcurrentIdenticalDeliveriesStoreOnce reproduces a reply that +// arrives over several paths at the same instant (the sweep saw list-agents +// duplicates 83 ns apart): exactly one copy may be stored. +func TestService_ConcurrentIdenticalDeliveriesStoreOnce(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + svc := NewService(ServiceConfig{InboxDir: tmp}) + frame := &Frame{Type: TypeText, Payload: []byte("answer"), MessageID: NewMessageID()} + + const paths = 8 + conns := make([]clientEnd, paths) + for i := range conns { + conns[i] = openServiceConn(t, svc, peerA) + } + var ( + wg sync.WaitGroup + mu sync.Mutex + duplicates int + errs []error + ) + start := make(chan struct{}) + for _, conn := range conns { + wg.Add(1) + go func(conn clientEnd) { + defer wg.Done() + <-start + res, err := sendAndAwaitAck(conn, frame) + mu.Lock() + defer mu.Unlock() + if err != nil { + errs = append(errs, err) + return + } + if res.Duplicate { + duplicates++ + } + }(conn) + } + close(start) + wg.Wait() + if len(errs) > 0 { + t.Fatalf("concurrent sends failed: %v", errs) + } + if duplicates != paths-1 { + t.Fatalf("duplicates = %d, want %d", duplicates, paths-1) + } + if n := len(inboxRecords(t, tmp)); n != 1 { + t.Fatalf("inbox records = %d, want exactly 1", n) + } +} + +// TestService_FailedDeliveryIsNotRemembered: a delivery that was not stored +// must not make its retry look like a duplicate. +func TestService_FailedDeliveryIsNotRemembered(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + inbox := filepath.Join(tmp, "inbox") + // A regular file where the inbox directory should be: MkdirAll fails. + if err := os.WriteFile(inbox, []byte("blocker"), 0600); err != nil { + t.Fatal(err) + } + svc := NewService(ServiceConfig{InboxDir: inbox}) + conn := openServiceConn(t, svc, peerA) + frame := &Frame{Type: TypeText, Payload: []byte("retry me"), MessageID: "r-1"} + + res, err := sendAndAwaitAck(conn, frame) + if !errors.Is(err, ErrRejected) || res == nil || res.Duplicate { + t.Fatalf("first attempt: res=%+v err=%v, want a rejection", res, err) + } + if err := os.Remove(inbox); err != nil { + t.Fatal(err) + } + res = mustSend(t, conn, frame) + if res.Duplicate { + t.Fatal("the retry of a failed delivery was treated as a duplicate") + } + if n := len(inboxRecords(t, inbox)); n != 1 { + t.Fatalf("inbox records = %d, want 1", n) + } +} + +func TestService_TaggedGovernedFrameKeepsCorrelation(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + governed, verifier := newGovernedTestFrame(t, &Frame{Type: TypeText, Payload: []byte("approved")}, decision.Allow, nil) + svc := NewService(ServiceConfig{InboxDir: tmp, RequireGoverned: true, GovernedVerifier: verifier}) + conn := openServiceConn(t, svc, peerA) + + envelope, err := EncodeGovernedFrame(governed) + if err != nil { + t.Fatal(err) + } + envelope.MessageID, envelope.ReplyTo = "gov-1", "req-7" + res := mustSend(t, conn, envelope) + if !res.Tagged || string(res.Ack.Payload) != "ACK TEXT 8 bytes" { + t.Fatalf("result=%+v ack=%q", res, res.Ack.Payload) + } + recs := inboxRecords(t, tmp) + if len(recs) != 1 || recs[0]["message_id"] != "gov-1" || recs[0]["reply_to"] != "req-7" || recs[0]["data"] != "approved" { + t.Fatalf("records = %v", recs) + } + // An identical re-delivery is acknowledged as a duplicate instead of + // tripping the governed replay guard with an error. + res = mustSend(t, conn, envelope) + if !res.Duplicate || string(res.Ack.Payload) != "ACK TEXT 8 bytes (duplicate)" { + t.Fatalf("re-delivery result=%+v ack=%q", res, res.Ack.Payload) + } + if n := len(inboxRecords(t, tmp)); n != 1 { + t.Fatalf("inbox records = %d, want 1", n) + } +} + +func TestService_TaggedTraceFrameKeepsCorrelation(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + conn := openServiceConn(t, NewService(ServiceConfig{InboxDir: tmp}), peerA) + payload := make([]byte, 12+2) + binary.BigEndian.PutUint32(payload[0:4], TypeText) + binary.BigEndian.PutUint64(payload[4:12], uint64(time.Now().UnixNano())) + copy(payload[12:], "hi") + res := mustSend(t, conn, &Frame{Type: TypeTrace, Payload: payload, MessageID: "trace-1"}) + if res.Ack.Type != TypeJSON { + t.Fatalf("trace ack type = %d", res.Ack.Type) + } + recs := inboxRecords(t, tmp) + if len(recs) != 1 || recs[0]["message_id"] != "trace-1" || recs[0]["data"] != "hi" { + t.Fatalf("records = %v", recs) + } +} + +func TestService_TaggedFileEventCarriesCorrelation(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + events := newCapturingEvents() + svc := NewService(ServiceConfig{ReceivedDir: tmp, InboxDir: t.TempDir()}) + svc.deps = coreapi.Deps{Events: events} + conn := openServiceConn(t, svc, peerA) + mustSend(t, conn, &Frame{Type: TypeFile, Filename: "r.txt", Payload: []byte("data"), MessageID: "f-1", ReplyTo: "req-2"}) + if res := mustSend(t, conn, &Frame{Type: TypeFile, Filename: "r.txt", Payload: []byte("data"), MessageID: "f-1", ReplyTo: "req-2"}); !res.Duplicate { + t.Fatal("identical tagged file re-delivery was stored again") + } + entries, _ := os.ReadDir(tmp) + if len(entries) != 1 { + t.Fatalf("received files = %d, want 1", len(entries)) + } + events.mu.Lock() + defer events.mu.Unlock() + found := false + for _, e := range events.published { + if e.topic == "file.received" { + found = e.payload["message_id"] == "f-1" && e.payload["reply_to"] == "req-2" + } + } + if !found { + t.Fatalf("file.received event lacks correlation: %v", events.published) + } +} + +// receiverGeneration is a dataexchange release line that predates +// TypeTagged, as emulated by oldReceiver. +type receiverGeneration int + +const ( + // genDropUnknown is v0.1.0 through v0.2.1 (web4 v1.12.x through + // v1.13.9): handleConn has no case for an unknown type, so the frame is + // not stored and is still acknowledged "ACK UNKNOWN() bytes". + genDropUnknown receiverGeneration = iota + // genRejectUnknown is v0.2.2 and later, before TypeTagged: an unknown + // type is answered "ERR UNKNOWN() save failed: unsupported frame type ". + genRejectUnknown + // genRejectUnknownGoverned is genRejectUnknown with RequireGoverned, + // which rejects every non-governed type before looking at it. + genRejectUnknownGoverned +) + +func (g receiverGeneration) String() string { + return [...]string{"v0.2.1-and-older", "v0.2.2+", "v0.2.2+/governed"}[g] +} + +var allOldReceiverGenerations = []receiverGeneration{genDropUnknown, genRejectUnknown, genRejectUnknownGoverned} + +// oldReceiver emulates a dataexchange receiver of generation gen, which +// predates TypeTagged: it reads raw frames and answers exactly as that +// release's service did. It returns the sender's end of the connection, +// the types of all frames received, and the payloads the receiver stored. +func oldReceiver(t *testing.T, gen receiverGeneration) (clientEnd, func() []uint32, func() [][]byte) { + t.Helper() + c2sR, c2sW := io.Pipe() + s2cR, s2cW := io.Pipe() + var ( + mu sync.Mutex + types []uint32 + stored [][]byte + ) + go func() { + defer s2cW.Close() + for { + var hdr [8]byte + if _, err := io.ReadFull(c2sR, hdr[:]); err != nil { + return + } + ftype := binary.BigEndian.Uint32(hdr[0:4]) + payload := make([]byte, binary.BigEndian.Uint32(hdr[4:8])) + if _, err := io.ReadFull(c2sR, payload); err != nil { + return + } + name, known := map[uint32]string{TypeText: "TEXT", TypeJSON: "JSON", TypeBinary: "BINARY"}[ftype] + if !known { + name = fmt.Sprintf("UNKNOWN(%d)", ftype) + } + ack := fmt.Sprintf("ACK %s %d bytes", name, len(payload)) + switch { + case gen == genRejectUnknownGoverned && ftype == TypeTagged: + ack = fmt.Sprintf("ERR %s save failed: unsigned legacy frame rejected by governed receiver", name) + case !known && gen != genDropUnknown: + ack = fmt.Sprintf("ERR %s save failed: unsupported frame type %d", name, ftype) + } + mu.Lock() + types = append(types, ftype) + if known { + stored = append(stored, payload) + } + mu.Unlock() + if err := WriteFrame(s2cW, &Frame{Type: TypeText, Payload: []byte(ack)}); err != nil { + return + } + } + }() + t.Cleanup(func() { _ = c2sW.Close(); _ = s2cR.Close() }) + seenTypes := func() []uint32 { mu.Lock(); defer mu.Unlock(); return append([]uint32(nil), types...) } + storedPayloads := func() [][]byte { mu.Lock(); defer mu.Unlock(); return append([][]byte(nil), stored...) } + return clientEnd{r: s2cR, w: c2sW}, seenTypes, storedPayloads +} + +// TestSend_FallsBackForOldReceiver: a sender using MessageID still reaches +// every receiver generation that predates TypeTagged; the message is stored +// exactly once, untagged. The v0.2.1-and-older case is review finding +// DX-REV-1: those receivers answer a tagged frame with a success ACK +// ("ACK UNKNOWN(10) ...") while dropping it, and Send used to report it as +// delivered with Tagged == true. +func TestSend_FallsBackForOldReceiver(t *testing.T) { + t.Parallel() + for _, gen := range allOldReceiverGenerations { + conn, types, stored := oldReceiver(t, gen) + res, err := sendAndAwaitAck(conn, &Frame{Type: TypeText, Payload: []byte("hello"), MessageID: "m-1", ReplyTo: "r-0"}) + if err != nil { + t.Fatalf("%v: send: %v", gen, err) + } + if res.Tagged || res.Duplicate || string(res.Ack.Payload) != "ACK TEXT 5 bytes" { + t.Fatalf("%v: result=%+v ack=%q", gen, res, res.Ack.Payload) + } + if got := types(); len(got) != 2 || got[0] != TypeTagged || got[1] != TypeText { + t.Fatalf("%v: old receiver saw types %v, want [TAGGED TEXT]", gen, got) + } + if got := stored(); len(got) != 1 || string(got[0]) != "hello" { + t.Fatalf("%v: old receiver stored %q, want exactly [hello]", gen, got) + } + } +} + +// TestSend_UntaggedFrameNeverRetried: a frame without metadata is sent once, +// whatever the receiver answers. +func TestSend_UntaggedFrameNeverRetried(t *testing.T) { + t.Parallel() + for _, gen := range allOldReceiverGenerations { + conn, types, _ := oldReceiver(t, gen) + res, err := sendAndAwaitAck(conn, &Frame{Type: 99, Payload: []byte("x")}) + if res == nil || (gen != genDropUnknown && !errors.Is(err, ErrRejected)) { + t.Fatalf("%v: res=%+v err=%v, want ErrRejected", gen, res, err) + } + if got := types(); len(got) != 1 { + t.Fatalf("%v: receiver saw %d frames, want 1", gen, len(got)) + } + } +} + +// TestReceiverPredatesTagged pins which acknowledgements trigger the +// untagged re-send. Only a receiver that does not know type 10 names it +// "UNKNOWN(10)"; a current receiver acknowledges the unwrapped inner type. +func TestReceiverPredatesTagged(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + ack *Frame + want bool + }{ + {&Frame{Type: TypeText, Payload: []byte("ACK UNKNOWN(10) 42 bytes")}, true}, + {&Frame{Type: TypeText, Payload: []byte("ERR UNKNOWN(10) save failed: unsupported frame type 10")}, true}, + {&Frame{Type: TypeText, Payload: []byte("ERR UNKNOWN(10) save failed: unsigned legacy frame rejected by governed receiver")}, true}, + {&Frame{Type: TypeText, Payload: []byte("ACK TEXT 5 bytes")}, false}, + {&Frame{Type: TypeText, Payload: []byte("ACK TEXT 5 bytes (duplicate)")}, false}, + {&Frame{Type: TypeText, Payload: []byte("ERR TEXT save failed: inbox byte budget exceeded")}, false}, + {&Frame{Type: TypeText, Payload: []byte("ACK UNKNOWN(11) 3 bytes")}, false}, + {&Frame{Type: TypeText, Payload: []byte("ACK UNKNOWN(100) 3 bytes")}, false}, + {&Frame{Type: TypeText, Payload: []byte("ERR UNKNOWN(11) save failed: unsupported frame type 11")}, false}, + {&Frame{Type: TypeJSON, Payload: []byte("ACK UNKNOWN(10) 42 bytes")}, false}, + {&Frame{Type: TypeText, Payload: nil}, false}, + } { + if got := receiverPredatesTagged(tc.ack); got != tc.want { + t.Errorf("receiverPredatesTagged(%d %q) = %v, want %v", tc.ack.Type, tc.ack.Payload, got, tc.want) + } + } +} + +// TestSend_CurrentReceiverKeepsTagsForEveryType: a current receiver never +// answers in a way that triggers the fallback, whatever the inner type. +func TestSend_CurrentReceiverKeepsTagsForEveryType(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + conn := openServiceConn(t, NewService(ServiceConfig{InboxDir: tmp, ReceivedDir: filepath.Join(tmp, "received")}), peerA) + for _, f := range []*Frame{ + {Type: TypeText, Payload: []byte("t")}, + {Type: TypeJSON, Payload: []byte(`{"a":1}`)}, + {Type: TypeBinary, Payload: []byte{0, 1, 2}}, + {Type: TypeFile, Filename: "f.txt", Payload: []byte("file")}, + } { + f.MessageID = NewMessageID() + res := mustSend(t, conn, f) + if !res.Tagged || receiverPredatesTagged(res.Ack) || !strings.HasPrefix(string(res.Ack.Payload), "ACK "+TypeName(f.Type)+" ") { + t.Fatalf("type %s: result=%+v ack=%q", TypeName(f.Type), res, res.Ack.Payload) + } + } + // An unknown inner type is rejected under its own name, not UNKNOWN(10), + // so it is not mistaken for an old receiver and re-sent. + res, err := sendAndAwaitAck(conn, &Frame{Type: 99, Payload: []byte("x"), MessageID: NewMessageID()}) + if !errors.Is(err, ErrRejected) || res == nil || !res.Tagged || !strings.HasPrefix(string(res.Ack.Payload), "ERR UNKNOWN(99) ") { + t.Fatalf("unknown inner type: res=%+v err=%v", res, err) + } +} + +// TestSend_NewReceiverRejectionIsNotRetried: a current receiver that rejects +// a tagged frame (here: unsigned under RequireGoverned) must not trigger the +// old-receiver fallback. +func TestSend_NewReceiverRejectionIsNotRetried(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + _, verifier := newGovernedTestFrame(t, &Frame{Type: TypeText, Payload: []byte("x")}, decision.Allow, nil) + conn := openServiceConn(t, NewService(ServiceConfig{InboxDir: tmp, RequireGoverned: true, GovernedVerifier: verifier}), peerA) + res, err := sendAndAwaitAck(conn, &Frame{Type: TypeText, Payload: []byte("unsigned"), MessageID: "m-1"}) + if !errors.Is(err, ErrRejected) || res == nil || !res.Tagged { + t.Fatalf("res=%+v err=%v", res, err) + } + if !strings.HasPrefix(string(res.Ack.Payload), "ERR TEXT save failed: unsigned legacy frame rejected") { + t.Fatalf("ack = %q", res.Ack.Payload) + } + if n := len(inboxRecords(t, tmp)); n != 0 { + t.Fatalf("inbox records = %d, want 0", n) + } +} + +func TestSend_NilFrame(t *testing.T) { + t.Parallel() + if _, err := sendAndAwaitAck(clientEnd{}, nil); err == nil { + t.Fatal("nil frame accepted") + } +} + +// smallCapChildEnv marks the child process of +// TestSend_TaggedFrameSizeEndToEndSmallCap. +const smallCapChildEnv = "DATAEXCHANGE_TEST_SMALL_CAP_CHILD" + +// TestSend_TaggedFrameSizeEndToEndSmallCap reproduces review finding +// DX-REV-3 against a real Service. MaxFrameSize is fixed at process start +// from PILOT_DATAEXCHANGE_MAX_FRAME, so the test re-runs itself in a child +// process with a 64 KiB cap. A frame of exactly the cap used to be sent +// tagged, over the cap: the receiver dropped the connection and Send +// returned a transport error with nothing stored. +func TestSend_TaggedFrameSizeEndToEndSmallCap(t *testing.T) { + if os.Getenv(smallCapChildEnv) != "1" { + t.Parallel() + cmd := exec.Command(os.Args[0], "-test.run=^TestSend_TaggedFrameSizeEndToEndSmallCap$", "-test.v", "-test.count=1") + cmd.Env = append(os.Environ(), smallCapChildEnv+"=1", "PILOT_DATAEXCHANGE_MAX_FRAME=65536") + out, err := cmd.CombinedOutput() + if err != nil || !strings.Contains(string(out), "--- PASS: TestSend_TaggedFrameSizeEndToEndSmallCap") { + t.Fatalf("child run failed (err %v):\n%s", err, out) + } + return + } + + if MaxFrameSize != 65536 { + t.Fatalf("MaxFrameSize = %d, want 65536 from PILOT_DATAEXCHANGE_MAX_FRAME", MaxFrameSize) + } + tmp := t.TempDir() + conn := openServiceConn(t, NewService(ServiceConfig{InboxDir: tmp}), peerA) + + // A frame at the cap fits only untagged, so it is delivered untagged. + atCap := &Frame{Type: TypeBinary, Payload: make([]byte, MaxFrameSize), MessageID: NewMessageID()} + res, err := sendAndAwaitAck(conn, atCap) + if err != nil { + t.Fatalf("frame at the cap: %v", err) + } + if want := fmt.Sprintf("ACK BINARY %d bytes", MaxFrameSize); res.Tagged || string(res.Ack.Payload) != want { + t.Fatalf("frame at the cap: result=%+v ack=%q, want untagged %q", res, res.Ack.Payload, want) + } + + // The largest payload that stays taggable with two maximal IDs is + // delivered tagged, on the same connection. + id := strings.Repeat("i", MaxMessageIDLen) + reply := strings.Repeat("r", MaxMessageIDLen) + fits := &Frame{Type: TypeBinary, Payload: make([]byte, int(MaxFrameSize)-MaxTaggedOverhead), MessageID: id, ReplyTo: reply} + res = mustSend(t, conn, fits) + if want := fmt.Sprintf("ACK BINARY %d bytes", len(fits.Payload)); !res.Tagged || string(res.Ack.Payload) != want { + t.Fatalf("largest taggable frame: result=%+v ack=%q, want tagged %q", res, res.Ack.Payload, want) + } + + recs := inboxRecords(t, tmp) + if len(recs) != 2 { + t.Fatalf("inbox records = %d, want 2", len(recs)) + } + if _, ok := recs[0]["message_id"]; ok || recs[0]["bytes"] != float64(MaxFrameSize) { + t.Fatalf("frame at the cap stored as %v, want %d bytes without message_id", recs[0]["bytes"], MaxFrameSize) + } + if recs[1]["message_id"] != id || recs[1]["reply_to"] != reply { + t.Fatalf("largest taggable frame stored with message_id=%v reply_to=%v", recs[1]["message_id"], recs[1]["reply_to"]) + } + + // A raw WriteFrame of the oversized tagged frame fails without writing. + var w countingWriter + if err := WriteFrame(&w, atCap); !errors.Is(err, ErrTaggedFrameTooLarge) || w.n != 0 { + t.Fatalf("raw WriteFrame: err=%v wrote %d bytes, want ErrTaggedFrameTooLarge and 0", err, w.n) + } +} + +// ---- deliveryDedupe unit tests ------------------------------------------ + +func testKey(b byte) dedupeKey { var k dedupeKey; k[0] = b; return k } + +func TestDeliveryDedupe_WaiterTakesOverFailedDelivery(t *testing.T) { + t.Parallel() + d := newDeliveryDedupe(time.Minute, 16) + first, _, ok := d.claim(context.Background(), testKey(1)) + if !ok || first.cache == nil { + t.Fatal("first claim was not granted") + } + type outcome struct { + claim dedupeClaim + dup string + } + waiter := make(chan outcome, 1) + go func() { + c, dup, _ := d.claim(context.Background(), testKey(1)) + waiter <- outcome{c, dup} + }() + select { + case <-waiter: + t.Fatal("second claim did not wait for the in-flight delivery") + case <-time.After(50 * time.Millisecond): + } + first.finish(false, "") + got := <-waiter + if got.claim.cache == nil { + t.Fatalf("after the first delivery failed the waiter got duplicate %q instead of the claim", got.dup) + } + got.claim.finish(true, "ACK TEXT 1 bytes") + + // Now remembered: the next claim is a duplicate carrying the first ACK. + c, dup, ok := d.claim(context.Background(), testKey(1)) + if !ok || c.cache != nil || dup != "ACK TEXT 1 bytes" { + t.Fatalf("claim after store = %+v %q %v", c, dup, ok) + } +} + +func TestDeliveryDedupe_WaiterSeesStoredDelivery(t *testing.T) { + t.Parallel() + d := newDeliveryDedupe(time.Minute, 16) + first, _, _ := d.claim(context.Background(), testKey(2)) + result := make(chan string, 1) + go func() { + _, dup, _ := d.claim(context.Background(), testKey(2)) + result <- dup + }() + time.Sleep(20 * time.Millisecond) + first.finish(true, "ACK JSON 3 bytes") + if got := <-result; got != "ACK JSON 3 bytes" { + t.Fatalf("waiter duplicate ack = %q", got) + } +} + +func TestDeliveryDedupe_WaiterHonoursContext(t *testing.T) { + t.Parallel() + d := newDeliveryDedupe(time.Minute, 16) + first, _, _ := d.claim(context.Background(), testKey(3)) + defer first.finish(false, "") + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + if _, _, ok := d.claim(ctx, testKey(3)); ok { + t.Fatal("claim returned ok after its context ended") + } +} + +func TestDeliveryDedupe_ExpiryAndCapacity(t *testing.T) { + t.Parallel() + now := time.Unix(1_000, 0) + d := newDeliveryDedupe(10*time.Second, 2) + d.now = func() time.Time { return now } + store := func(k byte) { + c, _, _ := d.claim(context.Background(), testKey(k)) + if c.cache == nil { + t.Fatalf("key %d unexpectedly remembered", k) + } + c.finish(true, "ACK") + } + remembered := func(k byte) bool { + c, _, _ := d.claim(context.Background(), testKey(k)) + if c.cache != nil { + c.finish(false, "") // not remembered: release the probe + return false + } + return true + } + + store(1) + store(2) + if !remembered(1) || !remembered(2) { + t.Fatal("stored keys not remembered") + } + store(3) // over capacity: the oldest (1) is forgotten early + if remembered(1) || !remembered(3) { + t.Fatal("capacity eviction did not drop the oldest entry") + } + now = now.Add(11 * time.Second) // past the window + if remembered(2) || remembered(3) { + t.Fatal("entries outlived the window") + } + if n := d.len(); n != 0 { + t.Fatalf("cache holds %d entries after expiry", n) + } + // A zero claim is inert. + var zero dedupeClaim + zero.finish(true, "ignored") +} + +// TestDeliveryDedupe_CompactsOrder keeps the FIFO bounded across many +// expiries. +func TestDeliveryDedupe_CompactsOrder(t *testing.T) { + t.Parallel() + now := time.Unix(1_000, 0) + d := newDeliveryDedupe(time.Second, 1<<20) + d.now = func() time.Time { return now } + for i := 0; i < 5000; i++ { + var k dedupeKey + binary.BigEndian.PutUint32(k[:4], uint32(i)) + c, _, _ := d.claim(context.Background(), k) + c.finish(true, "ACK") + now = now.Add(10 * time.Millisecond) // ~100 live at a time + } + d.mu.Lock() + defer d.mu.Unlock() + if live := len(d.order) - d.head; live > 200 { + t.Fatalf("%d live entries, want about 100", live) + } + if len(d.order) > 1000 { + t.Fatalf("order slice grew to %d; compaction is not running", len(d.order)) + } +} + +func TestDeliveryKeyFor_FieldBoundaries(t *testing.T) { + t.Parallel() + a := deliveryKeyFor(peerA, &Frame{Type: TypeText, MessageID: "ab", ReplyTo: "c", Payload: []byte("x")}) + b := deliveryKeyFor(peerA, &Frame{Type: TypeText, MessageID: "a", ReplyTo: "bc", Payload: []byte("x")}) + if a == b { + t.Fatal("shifting bytes between MessageID and ReplyTo produced the same key") + } + c := deliveryKeyFor(peerA, &Frame{Type: TypeFile, Filename: "n", Payload: []byte("x")}) + d := deliveryKeyFor(peerA, &Frame{Type: TypeFile, Filename: "", Payload: []byte("nx")}) + if c == d { + t.Fatal("shifting bytes between filename and payload produced the same key") + } +} + +func TestEffectiveDedupeWindow_Defaults(t *testing.T) { + t.Parallel() + for _, tc := range []struct{ in, want time.Duration }{ + {0, DefaultDedupeWindow}, + {-1, 0}, + {3 * time.Second, 3 * time.Second}, + } { + if got := (&Service{cfg: ServiceConfig{DedupeWindow: tc.in}}).effectiveDedupeWindow(); got != tc.want { + t.Errorf("DedupeWindow %v ⇒ %v, want %v", tc.in, got, tc.want) + } + } +} diff --git a/zz_fuzz_frames_test.go b/zz_fuzz_frames_test.go index fe2af73..67b31e2 100644 --- a/zz_fuzz_frames_test.go +++ b/zz_fuzz_frames_test.go @@ -43,6 +43,13 @@ func FuzzDataExchangeRoundTrip(f *testing.F) { if len(payload) > 1<<20 { // keep manageable payload = payload[:1<<20] } + if ftype == dataexchange.TypeTagged { + // TypeTagged is produced by WriteFrame from MessageID/ReplyTo and + // unwrapped by ReadFrame, so a raw payload of that type does not + // round-trip; FuzzTaggedFrameRoundTrip / FuzzTaggedFrameRead + // cover it. + return + } frame := &dataexchange.Frame{Type: ftype, Payload: payload} var buf bytes.Buffer diff --git a/zz_harden_test.go b/zz_harden_test.go index 35163e0..e47df93 100644 --- a/zz_harden_test.go +++ b/zz_harden_test.go @@ -193,34 +193,27 @@ func TestSaveInboxMessage_BinaryLossless(t *testing.T) { func TestSaveInboxMessage_ByteCapEnforced(t *testing.T) { t.Parallel() tmp := t.TempDir() - // Small cap so a couple of writes blow past it. Each message is the - // payload + ~256 estimated JSON overhead, so cap at ~1 KiB. + // Small cap so a couple of writes blow past it: each message is the + // 400-byte payload plus its JSON envelope, so the cap holds ~2 of them. s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: 1024}) from := protocol.Addr{Node: 1} payload := bytes.Repeat([]byte("x"), 400) // valid UTF-8 - // First write fits. - if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: payload}, from); err != nil { - t.Fatalf("first write should fit: %v", err) - } - // Keep writing; the cap must eventually reject (after eviction can no - // longer make room, which here happens because every file is ~same size - // and the cap holds only ~1-2 of them). - rejected := false + // Keep writing well past the cap. The oldest messages must be evicted to + // make room (FIFO), so every write succeeds and the cap always holds. for i := 0; i < 20; i++ { if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: payload}, from); err != nil { - rejected = true - break + t.Fatalf("write %d: the byte cap should evict old messages, not reject new ones: %v", i, err) + } + total, _ := inboxTotalBytes(tmp) + if total > 1024 { + t.Fatalf("after write %d the inbox holds %d bytes, over the 1024-byte cap", i, total) } } - if !rejected { - t.Fatal("expected the inbox byte cap to reject a write once full") - } - // On-disk total must never have exceeded the cap by more than one message. - total, _ := inboxTotalBytes(tmp) - if total > 1024+int64(len(payload))+512 { - t.Fatalf("inbox grew to %d bytes, well past the 1024 cap", total) + entries, _ := os.ReadDir(tmp) + if len(entries) == 0 { + t.Fatal("the byte cap evicted everything; it must keep the newest messages") } } diff --git a/zz_inbox_budget_test.go b/zz_inbox_budget_test.go new file mode 100644 index 0000000..8543b41 --- /dev/null +++ b/zz_inbox_budget_test.go @@ -0,0 +1,518 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !no_dataexchange +// +build !no_dataexchange + +package dataexchange + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// seedSparseInbox writes n files of size bytes each into dir with mtimes an +// hour or more in the past (oldest first). The files are sparse, so a test +// can exceed the 256 MiB default cap without using real disk space. +func seedSparseInbox(t *testing.T, dir string, n int, size int64) { + t.Helper() + base := time.Now().Add(-time.Hour) + for i := 0; i < n; i++ { + p := filepath.Join(dir, fmt.Sprintf("OLD-%05d.json", i)) + f, err := os.Create(p) + if err != nil { + t.Fatalf("create %s: %v", p, err) + } + if err := f.Truncate(size); err != nil { + t.Fatalf("truncate %s: %v", p, err) + } + _ = f.Close() + mt := base.Add(time.Duration(i) * time.Second) + if err := os.Chtimes(p, mt, mt); err != nil { + t.Fatalf("chtimes %s: %v", p, err) + } + } +} + +func countPrefix(t *testing.T, dir, prefix string) int { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("readdir: %v", err) + } + n := 0 + for _, e := range entries { + if strings.HasPrefix(e.Name(), prefix) { + n++ + } + } + return n +} + +// TestInboxByteCap_DefaultConfigEvictsOldestNotEverything is the regression +// test for the sweep finding (HYG-01): with the DEFAULT config +// (InboxMaxBytes == 0 ⇒ 256 MiB) the byte-cap evictor compared against the +// raw config value 0 and deleted every inbox file, including the replies a +// `send-message --wait` was about to read. +func TestInboxByteCap_DefaultConfigEvictsOldestNotEverything(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + s := NewService(ServiceConfig{InboxDir: tmp}) // default caps + if got := s.effectiveInboxMaxBytes(); got != DefaultInboxMaxBytes { + t.Fatalf("effective cap = %d, want the default %d", got, DefaultInboxMaxBytes) + } + + // The sweep's reproduction: 257 MiB of old messages (already past the + // 256 MiB default), five recent replies, then one incoming message. + const mib = int64(1 << 20) + seedSparseInbox(t, tmp, 257, mib) + for i := 0; i < 5; i++ { + p := filepath.Join(tmp, fmt.Sprintf("RECENT-%d.json", i)) + if err := os.WriteFile(p, []byte(fmt.Sprintf(`{"data":"recent reply %d"}`, i)), 0600); err != nil { + t.Fatal(err) + } + } + from := protocol.Addr{Network: 1, Node: 0x2} + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: []byte("newest")}, from); err != nil { + t.Fatalf("save that trips the cap: %v", err) + } + + if got := countPrefix(t, tmp, "RECENT-"); got != 5 { + t.Fatalf("recent replies surviving eviction = %d/5 (the old evictor left 0/5)", got) + } + if got := countPrefix(t, tmp, "TEXT-"); got != 1 { + t.Fatalf("the incoming message was not stored (TEXT files = %d)", got) + } + old := countPrefix(t, tmp, "OLD-") + if old == 0 { + t.Fatal("the default byte cap wiped every old message; it must only evict the oldest overflow") + } + total, err := inboxTotalBytes(tmp) + if err != nil { + t.Fatal(err) + } + if total > DefaultInboxMaxBytes { + t.Fatalf("inbox holds %d bytes, over the %d default cap", total, DefaultInboxMaxBytes) + } + // One pass trims to the low-water mark (90%), not just under the cap. + if total > inboxLowWater(DefaultInboxMaxBytes) { + t.Fatalf("inbox holds %d bytes after eviction, want <= low-water %d", total, inboxLowWater(DefaultInboxMaxBytes)) + } + // ...and no further than that: roughly 10% of the cap plus the 1 MiB of + // overshoot is evicted, not the whole inbox. + if evicted := 257 - old; evicted > 30 { + t.Fatalf("evicted %d of 257 old messages, want only the ~27 needed to reach the low-water mark", evicted) + } +} + +// TestEvictInboxOverflow_DefaultConfigAppliesByteCap covers the periodic +// eviction pass with the default config: it used to skip the byte cap +// entirely whenever InboxMaxBytes was 0. +func TestEvictInboxOverflow_DefaultConfigAppliesByteCap(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + s := NewService(ServiceConfig{InboxDir: tmp}) + seedSparseInbox(t, tmp, 300, 1<<20) // 300 MiB, under the 10k file cap + + s.evictInboxOverflow(tmp) + + total, err := inboxTotalBytes(tmp) + if err != nil { + t.Fatal(err) + } + if total > inboxLowWater(DefaultInboxMaxBytes) { + t.Fatalf("after the periodic pass the inbox holds %d bytes, want <= %d", total, inboxLowWater(DefaultInboxMaxBytes)) + } + remaining := countPrefix(t, tmp, "OLD-") + if remaining == 0 { + t.Fatal("periodic eviction wiped the inbox") + } + // The newest file must be among the survivors. + if _, err := os.Stat(filepath.Join(tmp, "OLD-00299.json")); err != nil { + t.Fatalf("newest message was evicted: %v", err) + } + if s.inbox.onDisk != total { + t.Fatalf("running total %d not re-seeded from disk (%d)", s.inbox.onDisk, total) + } +} + +// TestInboxByteCap_NoDirectoryScanPerMessage checks the running byte +// counter: under the cap, saves never list the inbox directory after the +// first one (the old code listed and stat'ed every file on every message). +func TestInboxByteCap_NoDirectoryScanPerMessage(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + seedSparseInbox(t, tmp, 50, 1024) + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: 1 << 30}) + from := protocol.Addr{Node: 7} + // Stay below inboxEvictCheckEvery so the periodic pass does not run. + for i := 0; i < inboxEvictCheckEvery-1; i++ { + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: []byte("x")}, from); err != nil { + t.Fatalf("save %d: %v", i, err) + } + } + if s.inbox.scans != 1 { + t.Fatalf("inbox directory scanned %d times for %d saves, want 1 (seed only)", s.inbox.scans, inboxEvictCheckEvery-1) + } + total, _ := inboxTotalBytes(tmp) + if s.inbox.onDisk != total || s.inbox.pending != 0 { + t.Fatalf("running total = %d (pending %d), disk = %d", s.inbox.onDisk, s.inbox.pending, total) + } +} + +// TestInboxByteCap_StaleCounterRescansBeforeEvicting: other processes delete +// inbox files (pilotctl inbox --clear). The running total then overstates +// usage; the service must re-read the directory instead of evicting the +// messages that remain. +func TestInboxByteCap_StaleCounterRescansBeforeEvicting(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: 4096}) + from := protocol.Addr{Node: 8} + payload := []byte(strings.Repeat("p", 200)) + for i := 0; i < 12; i++ { + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: payload}, from); err != nil { + t.Fatalf("fill %d: %v", i, err) + } + } + // Another process clears the inbox behind the service's back. + entries, _ := os.ReadDir(tmp) + for _, e := range entries { + _ = os.Remove(filepath.Join(tmp, e.Name())) + } + // These writes would overflow the stale counter; a fresh scan shows an + // empty inbox, so none of them may evict another. + for i := 0; i < 5; i++ { + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: payload}, from); err != nil { + t.Fatalf("post-clear save %d: %v", i, err) + } + if got := countPrefix(t, tmp, "TEXT-"); got != i+1 { + t.Fatalf("after post-clear save %d the inbox has %d files, want %d (nothing evicted)", i, got, i+1) + } + } +} + +// TestInboxByteCap_LowWaterHysteresis: an eviction frees ~10% of the cap, so +// the following writes do not each pay for another eviction pass. +func TestInboxByteCap_LowWaterHysteresis(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + const capBytes = 20000 + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: capBytes}) + from := protocol.Addr{Node: 9} + payload := []byte(strings.Repeat("h", 100)) + + var afterEviction int + for i := 0; i < 500; i++ { + before := countPrefix(t, tmp, "TEXT-") + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: payload}, from); err != nil { + t.Fatalf("save %d: %v", i, err) + } + if countPrefix(t, tmp, "TEXT-") < before+1 { + afterEviction = i + break + } + } + if afterEviction == 0 { + t.Fatal("the byte cap never evicted") + } + total, _ := inboxTotalBytes(tmp) + if total > inboxLowWater(capBytes) { + t.Fatalf("after eviction the inbox holds %d bytes, want <= low-water %d", total, inboxLowWater(capBytes)) + } + // The next several writes fit without another eviction. + for i := 0; i < 5; i++ { + before := countPrefix(t, tmp, "TEXT-") + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: payload}, from); err != nil { + t.Fatalf("post-eviction save %d: %v", i, err) + } + if got := countPrefix(t, tmp, "TEXT-"); got != before+1 { + t.Fatalf("post-eviction save %d evicted again (%d -> %d files)", i, before, got) + } + } +} + +// TestInboxByteCap_ConcurrentWritersHoldCap runs saves from many goroutines +// (one per connection in production) and checks the cap and the running +// total stay consistent. Run with -race. +func TestInboxByteCap_ConcurrentWritersHoldCap(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + const capBytes = 16384 + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: capBytes}) + payload := []byte(strings.Repeat("c", 300)) + + var wg sync.WaitGroup + errs := make(chan error, 8*25) + for w := 0; w < 8; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + from := protocol.Addr{Node: uint32(100 + w)} + for i := 0; i < 25; i++ { + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: payload}, from); err != nil { + errs <- err + } + } + }(w) + } + wg.Wait() + close(errs) + for err := range errs { + t.Fatalf("concurrent save failed: %v", err) + } + total, _ := inboxTotalBytes(tmp) + if total > capBytes { + t.Fatalf("inbox holds %d bytes, over the %d cap", total, capBytes) + } + if s.inbox.pending != 0 { + t.Fatalf("pending = %d after all writes finished, want 0", s.inbox.pending) + } + if s.inbox.onDisk < total { + t.Fatalf("running total %d understates disk usage %d", s.inbox.onDisk, total) + } +} + +// TestInboxByteCap_RollbackReleasesBytes: a governed delivery that is +// rolled back (receipt failure) must hand its bytes back to the budget. +func TestInboxByteCap_RollbackReleasesBytes(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: 1 << 20}) + delivery, err := s.prepareInboxMessage(&Frame{Type: TypeText, Payload: []byte("to be rolled back")}, protocol.Addr{Node: 3}, nil) + if err != nil { + t.Fatalf("prepare: %v", err) + } + if s.inbox.onDisk == 0 { + t.Fatal("running total did not count the staged message") + } + if err := delivery.rollback(); err != nil { + t.Fatalf("rollback: %v", err) + } + if s.inbox.onDisk != 0 || s.inbox.pending != 0 { + t.Fatalf("after rollback onDisk=%d pending=%d, want 0/0", s.inbox.onDisk, s.inbox.pending) + } +} + +// TestInboxByteCap_DisabledSkipsAccounting: a negative InboxMaxBytes turns +// the byte cap off, and with it every scan and counter update. +func TestInboxByteCap_DisabledSkipsAccounting(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: -1}) + for i := 0; i < 10; i++ { + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: []byte(strings.Repeat("d", 1000))}, protocol.Addr{Node: 4}); err != nil { + t.Fatalf("save %d: %v", i, err) + } + } + if s.inbox.scans != 0 || s.inbox.onDisk != 0 { + t.Fatalf("disabled byte cap still accounted: scans=%d onDisk=%d", s.inbox.scans, s.inbox.onDisk) + } + if got := countPrefix(t, tmp, "TEXT-"); got != 10 { + t.Fatalf("files = %d, want 10", got) + } +} + +func TestInboxLowWater(t *testing.T) { + t.Parallel() + for _, tc := range []struct{ max, want int64 }{ + {100, 90}, + {200, 180}, + {1000, 900}, + {DefaultInboxMaxBytes, 241591910}, + {1<<62 + 7, 4150517416584649119}, // no overflow near the int64 limit + } { + if got := inboxLowWater(tc.max); got != tc.want { + t.Errorf("inboxLowWater(%d) = %d, want %d", tc.max, got, tc.want) + } + } +} + +// TestInboxByteCap_UnlistableInboxIsBestEffort keeps the historical +// behaviour: failing to list the inbox does not block delivery. +func TestInboxByteCap_UnlistableInboxIsBestEffort(t *testing.T) { + t.Parallel() + s := NewService(ServiceConfig{InboxMaxBytes: 1 << 20}) + missing := filepath.Join(t.TempDir(), "does-not-exist") + path := filepath.Join(missing, "TEXT-x.json") + if err := s.reserveInbox(missing, path, 100, 1<<20); err != nil { + t.Fatalf("reserve with an unlistable inbox: %v", err) + } + if s.inbox.pending != 100 { + t.Fatalf("pending = %d, want 100", s.inbox.pending) + } + s.inboxWriteDone(path, 100, false) + if s.inbox.pending != 0 || s.inbox.onDisk != 0 || len(s.inbox.inflight) != 0 { + t.Fatalf("after a failed write pending=%d onDisk=%d inflight=%d, want 0/0/0", s.inbox.pending, s.inbox.onDisk, len(s.inbox.inflight)) + } +} + +// fileWithMarker reports whether some file in dir contains marker. +func fileWithMarker(t *testing.T, dir, marker string) bool { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("readdir: %v", err) + } + for _, e := range entries { + raw, err := os.ReadFile(filepath.Join(dir, e.Name())) + if err == nil && strings.Contains(string(raw), marker) { + return true + } + } + return false +} + +// TestInboxByteCap_InflightWriteNotEvicted is the regression test for review +// finding DX-REV-2. Writer A is admitted and its file is fully written, but +// its reservation is not settled yet (the window between os.WriteFile and +// the deferred inboxWriteDone). Writer B then needs an eviction. B used to +// count A's bytes twice (in the fresh scan and in pending), evict down past +// the old file and delete A's file too, although A had been admitted and +// was about to be ACKed, and A + B fit under the cap. +func TestInboxByteCap_InflightWriteNotEvicted(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + const capBytes = 1000 + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: capBytes}) + oldPath := filepath.Join(tmp, "OLD.json") + if err := os.WriteFile(oldPath, []byte(strings.Repeat("o", 300)), 0600); err != nil { + t.Fatal(err) + } + mt := time.Now().Add(-time.Hour) + if err := os.Chtimes(oldPath, mt, mt); err != nil { + t.Fatal(err) + } + + // Hold writer A right after its file is on disk. + aWritten := make(chan string, 1) + releaseA := make(chan struct{}) + release := sync.OnceFunc(func() { close(releaseA) }) + defer release() + var writes atomic.Int32 + s.afterInboxWrite = func(path string) { + if writes.Add(1) == 1 { // writer A only + aWritten <- path + <-releaseA + } + } + from := protocol.Addr{Network: 1, Node: 0x2} + errA := make(chan error, 1) + go func() { + errA <- s.saveInboxMessage(&Frame{Type: TypeText, Payload: []byte("MARKER-A" + strings.Repeat("a", 292))}, from) + }() + aPath := <-aWritten + + if err := s.saveInboxMessage(&Frame{Type: TypeText, Payload: []byte("MARKER-B" + strings.Repeat("b", 292))}, from); err != nil { + t.Fatalf("save B: %v", err) + } + if _, err := os.Stat(aPath); err != nil { + t.Fatalf("B's eviction deleted A's file, which was admitted and fully written: %v", err) + } + release() + if err := <-errA; err != nil { + t.Fatalf("save A: %v", err) + } + + aInfo, err := os.Stat(aPath) + if err != nil { + t.Fatalf("A's file is gone after A was acknowledged: %v", err) + } + total, err := inboxTotalBytes(tmp) + if err != nil { + t.Fatal(err) + } + bSize := total - aInfo.Size() + if _, err := os.Stat(oldPath); err == nil { + bSize -= 300 + } + // The premise: OLD + A + B does not fit, A + B fits under the low-water + // mark, so exactly OLD has to go. + if 300+aInfo.Size()+bSize <= capBytes || aInfo.Size()+bSize > inboxLowWater(capBytes) { + t.Fatalf("test premise broken: A=%d B=%d bytes", aInfo.Size(), bSize) + } + if _, err := os.Stat(oldPath); !os.IsNotExist(err) { + t.Fatalf("the oldest message was not evicted to make room (stat err %v)", err) + } + if !fileWithMarker(t, tmp, "MARKER-A") || !fileWithMarker(t, tmp, "MARKER-B") { + t.Fatal("A or B missing from the inbox") + } + if total > capBytes { + t.Fatalf("inbox holds %d bytes, over the %d cap", total, capBytes) + } + // The running total counts each file once. + if s.inbox.pending != 0 || s.inbox.onDisk != total || len(s.inbox.inflight) != 0 { + t.Fatalf("budget pending=%d onDisk=%d inflight=%d, want 0/%d/0", s.inbox.pending, s.inbox.onDisk, len(s.inbox.inflight), total) + } +} + +// TestInboxByteCap_RescanDoesNotDoubleCountInflight: a rescan while a write +// is in flight must not add the in-flight file to onDisk, or the running +// total overstates the inbox once the write settles. +func TestInboxByteCap_RescanDoesNotDoubleCountInflight(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: 1 << 20}) + path := filepath.Join(tmp, "TEXT-inflight.json") + if err := s.reserveInbox(tmp, path, 400, 1<<20); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, make([]byte, 400), 0600); err != nil { + t.Fatal(err) + } + s.evictInboxOverflow(tmp) // rescans while the write is in flight + if s.inbox.onDisk != 0 { + t.Fatalf("onDisk = %d while the only file is in flight, want 0", s.inbox.onDisk) + } + s.inboxWriteDone(path, 400, true) + if s.inbox.onDisk != 400 || s.inbox.pending != 0 { + t.Fatalf("after settling onDisk=%d pending=%d, want 400/0", s.inbox.onDisk, s.inbox.pending) + } +} + +// TestEvictInboxOverflow_SkipsInflightWrite: the periodic pass must not +// delete a file whose write is in flight either. It trims the settled files +// so that they fit next to the in-flight bytes. +func TestEvictInboxOverflow_SkipsInflightWrite(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + const capBytes = 1000 + s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: capBytes}) + path := filepath.Join(tmp, "TEXT-inflight.json") + if err := s.reserveInbox(tmp, path, 950, capBytes); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, make([]byte, 950), 0600); err != nil { + t.Fatal(err) + } + // Another process drops a file in while the write is in flight, and the + // in-flight file is (for this test) older than it. + other := filepath.Join(tmp, "OTHER.json") + if err := os.WriteFile(other, make([]byte, 100), 0600); err != nil { + t.Fatal(err) + } + older := time.Now().Add(-time.Hour) + if err := os.Chtimes(path, older, older); err != nil { + t.Fatal(err) + } + + s.evictInboxOverflow(tmp) + + if _, err := os.Stat(path); err != nil { + t.Fatalf("periodic eviction deleted an in-flight write: %v", err) + } + if _, err := os.Stat(other); !os.IsNotExist(err) { + t.Fatalf("settled file not trimmed although settled + in-flight is over the cap (stat err %v)", err) + } + s.inboxWriteDone(path, 950, true) + if total, _ := inboxTotalBytes(tmp); s.inbox.onDisk != total || total > capBytes { + t.Fatalf("onDisk=%d disk=%d cap=%d", s.inbox.onDisk, total, capBytes) + } +} diff --git a/zz_service_errors_test.go b/zz_service_errors_test.go index 12a9a06..e4d8b9d 100644 --- a/zz_service_errors_test.go +++ b/zz_service_errors_test.go @@ -426,7 +426,8 @@ func (e *capturingEvents) Subscribe(pattern string) (<-chan coreapi.Event, func( // ---- PILOT-276: inbox byte-budget cap ---------------------------------- // TestSaveInboxMessage_ByteBudgetEnforced verifies that when InboxMaxBytes is -// set, saveInboxMessage rejects writes that would exceed the budget. +// set, saveInboxMessage keeps the inbox within the budget by evicting the +// oldest messages, and rejects only a message that cannot fit at all. func TestSaveInboxMessage_ByteBudgetEnforced(t *testing.T) { t.Parallel() tmp := t.TempDir() @@ -438,22 +439,29 @@ func TestSaveInboxMessage_ByteBudgetEnforced(t *testing.T) { t.Fatalf("first save under budget must succeed: %v", err) } - // Second message — the inbox JSON overhead + data should push over 512. - // Force the test by writing a payload that fills the rest of the budget. - // After the first save, let's check how many files exist. - entries, _ := os.ReadDir(tmp) - t.Logf("entries after first save: %d", len(entries)) - - // Save enough messages to exceed 512 bytes total. + // Save enough messages to exceed 512 bytes several times over: all are + // accepted and the total never exceeds the budget. for i := 0; i < 20; i++ { frame := &Frame{Type: TypeText, Payload: []byte(strings.Repeat("x", 50))} - err := s.saveInboxMessage(frame, protocol.Addr{Node: 2}) - if err != nil { - t.Logf("saveInboxMessage #%d failed as expected: %v", i+2, err) - return // success — budget enforced + if err := s.saveInboxMessage(frame, protocol.Addr{Node: 2}); err != nil { + t.Fatalf("save #%d: %v", i+2, err) + } + if total, _ := inboxTotalBytes(tmp); total > 512 { + t.Fatalf("save #%d: inbox holds %d bytes, over the 512-byte budget", i+2, total) } } - t.Error("expected saveInboxMessage to eventually reject when over byte budget") + + // A message larger than the whole budget is rejected, and nothing is + // evicted on its behalf. + before, _ := os.ReadDir(tmp) + big := &Frame{Type: TypeText, Payload: []byte(strings.Repeat("z", 600))} + if err := s.saveInboxMessage(big, protocol.Addr{Node: 2}); err == nil { + t.Fatal("a message larger than the byte budget must be rejected") + } + after, _ := os.ReadDir(tmp) + if len(after) != len(before) { + t.Fatalf("rejecting an oversized message changed the inbox: %d files before, %d after", len(before), len(after)) + } } // TestSaveInboxMessage_NoByteBudget_Unbounded documents that with diff --git a/zz_tagged_test.go b/zz_tagged_test.go new file mode 100644 index 0000000..db344da --- /dev/null +++ b/zz_tagged_test.go @@ -0,0 +1,417 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package dataexchange + +import ( + "bytes" + "encoding/binary" + "errors" + "fmt" + "io" + "strings" + "testing" + "time" +) + +// rawFrame builds wire bytes by hand, bypassing WriteFrame's validation. +func rawFrame(ftype uint32, payload []byte) []byte { + out := make([]byte, 8+len(payload)) + binary.BigEndian.PutUint32(out[0:4], ftype) + binary.BigEndian.PutUint32(out[4:8], uint32(len(payload))) + copy(out[8:], payload) + return out +} + +// rawTaggedPayload builds a TypeTagged payload from a literal header. +func rawTaggedPayload(header string, innerType uint32, inner []byte) []byte { + out := make([]byte, 2+len(header)+4+len(inner)) + binary.BigEndian.PutUint16(out[0:2], uint16(len(header))) + copy(out[2:], header) + binary.BigEndian.PutUint32(out[2+len(header):], innerType) + copy(out[2+len(header)+4:], inner) + return out +} + +// TestWriteFrame_UntaggedWireFormatUnchanged pins the bytes of a frame +// without correlation metadata: older peers must see exactly the original +// format. +func TestWriteFrame_UntaggedWireFormatUnchanged(t *testing.T) { + t.Parallel() + var buf bytes.Buffer + if err := WriteFrame(&buf, &Frame{Type: TypeText, Payload: []byte("hi")}); err != nil { + t.Fatal(err) + } + want := []byte{0, 0, 0, 1, 0, 0, 0, 2, 'h', 'i'} + if !bytes.Equal(buf.Bytes(), want) { + t.Fatalf("untagged text frame = %v, want %v", buf.Bytes(), want) + } + + buf.Reset() + if err := WriteFrame(&buf, &Frame{Type: TypeFile, Filename: "a", Payload: []byte("z")}); err != nil { + t.Fatal(err) + } + want = []byte{0, 0, 0, 4, 0, 0, 0, 4, 0, 1, 'a', 'z'} + if !bytes.Equal(buf.Bytes(), want) { + t.Fatalf("untagged file frame = %v, want %v", buf.Bytes(), want) + } +} + +func TestTaggedFrame_RoundTrip(t *testing.T) { + t.Parallel() + id := NewMessageID() + for _, tc := range []Frame{ + {Type: TypeText, Payload: []byte("what is the weather"), MessageID: id}, + {Type: TypeJSON, Payload: []byte(`{"q":1}`), ReplyTo: id}, + {Type: TypeBinary, Payload: []byte{0, 0xFF, 7}, MessageID: "reply-1", ReplyTo: id}, + {Type: TypeFile, Filename: "report.csv", Payload: []byte("a,b"), MessageID: id}, + {Type: TypeText, Payload: []byte{}, MessageID: id}, + } { + tc := tc + var buf bytes.Buffer + if err := WriteFrame(&buf, &tc); err != nil { + t.Fatalf("WriteFrame(%+v): %v", tc, err) + } + if got := binary.BigEndian.Uint32(buf.Bytes()[0:4]); got != TypeTagged { + t.Fatalf("on-wire type = %d, want TypeTagged", got) + } + got, err := ReadFrame(&buf) + if err != nil { + t.Fatalf("ReadFrame: %v", err) + } + if got.Type != tc.Type || !bytes.Equal(got.Payload, tc.Payload) || got.Filename != tc.Filename || + got.MessageID != tc.MessageID || got.ReplyTo != tc.ReplyTo { + t.Fatalf("round trip = %+v, want %+v", got, tc) + } + } +} + +func TestTaggedFrame_WriteRejectsInvalidIDs(t *testing.T) { + t.Parallel() + for _, f := range []Frame{ + {Type: TypeText, MessageID: strings.Repeat("a", MaxMessageIDLen+1)}, + {Type: TypeText, MessageID: "has space"}, + {Type: TypeText, ReplyTo: "quote\""}, + {Type: TypeText, ReplyTo: "slash/../x"}, + {Type: TypeText, MessageID: "ünïcode"}, + {Type: TypeTagged, MessageID: "nested"}, + {Type: TypeFile, Filename: strings.Repeat("n", maxFilenameLen+1), MessageID: "ok"}, + } { + f := f + var buf bytes.Buffer + if err := WriteFrame(&buf, &f); err == nil { + t.Errorf("WriteFrame(%+v) accepted an invalid frame", f) + } + if buf.Len() != 0 { + t.Errorf("WriteFrame(%+v) wrote %d bytes before failing", f, buf.Len()) + } + } + // The longest valid ID is accepted. + var buf bytes.Buffer + if err := WriteFrame(&buf, &Frame{Type: TypeText, MessageID: strings.Repeat("A", MaxMessageIDLen)}); err != nil { + t.Fatalf("max-length ID rejected: %v", err) + } +} + +func TestTaggedFrame_ReadRejectsMalformed(t *testing.T) { + t.Parallel() + cases := map[string][]byte{ + "empty payload": {}, + "one byte": {0}, + "zero header length": rawTaggedPayload("", TypeText, []byte("x")), + "truncated inner type": append([]byte{0, 2}, []byte("{}")...), + "header past end": {0, 50, '{', '}'}, + "not json": rawTaggedPayload("nope", TypeText, []byte("x")), + "wrong json type": rawTaggedPayload(`{"message_id":7}`, TypeText, []byte("x")), + "invalid id chars": rawTaggedPayload(`{"message_id":"a b"}`, TypeText, []byte("x")), + "id too long": rawTaggedPayload(`{"reply_to":"`+strings.Repeat("r", MaxMessageIDLen+1)+`"}`, TypeText, []byte("x")), + "nested tagged": rawTaggedPayload(`{"message_id":"a"}`, TypeTagged, rawTaggedPayload(`{}`, TypeText, nil)), + "traversal filename": rawTaggedPayload(`{"message_id":"a"}`, TypeFile, append([]byte{0, 4}, []byte("../x")...)), + } + huge := make([]byte, 2+maxTaggedHeaderLen+1+4) + binary.BigEndian.PutUint16(huge[0:2], maxTaggedHeaderLen+1) + cases["header too long"] = huge + + for name, payload := range cases { + if _, err := ReadFrame(bytes.NewReader(rawFrame(TypeTagged, payload))); err == nil { + t.Errorf("%s: ReadFrame accepted a malformed tagged frame", name) + } + } +} + +// TestTaggedFrame_ReadIgnoresUnknownHeaderKeys: later versions may add header +// fields; this version must still deliver the frame. +func TestTaggedFrame_ReadIgnoresUnknownHeaderKeys(t *testing.T) { + t.Parallel() + payload := rawTaggedPayload(`{"message_id":"m-1","trace_parent":"00-abc","priority":3}`, TypeJSON, []byte(`{"a":1}`)) + f, err := ReadFrame(bytes.NewReader(rawFrame(TypeTagged, payload))) + if err != nil { + t.Fatalf("ReadFrame: %v", err) + } + if f.Type != TypeJSON || f.MessageID != "m-1" || f.ReplyTo != "" || string(f.Payload) != `{"a":1}` { + t.Fatalf("frame = %+v", f) + } + // An empty header object is a tagged frame without metadata. + f, err = ReadFrame(bytes.NewReader(rawFrame(TypeTagged, rawTaggedPayload(`{}`, TypeText, []byte("x"))))) + if err != nil || f.Type != TypeText || f.MessageID != "" || string(f.Payload) != "x" { + t.Fatalf("empty header: frame=%+v err=%v", f, err) + } +} + +func TestNewMessageID(t *testing.T) { + t.Parallel() + seen := make(map[string]bool) + for i := 0; i < 1000; i++ { + id := NewMessageID() + if len(id) != 32 || !ValidMessageID(id) { + t.Fatalf("NewMessageID() = %q, want 32 valid hex characters", id) + } + if seen[id] { + t.Fatalf("NewMessageID repeated %q", id) + } + seen[id] = true + } +} + +func TestValidMessageID(t *testing.T) { + t.Parallel() + for id, want := range map[string]bool{ + "": false, + "abc": true, + "01J9Z8X5Q3R4T6Y7U8I9O0P1A2": true, // ULID + "123e4567-e89b-12d3-a456-426614174000": true, // UUID + "req:42.retry_1": true, + "a b": false, + "a\nb": false, + "a/b": false, + strings.Repeat("x", MaxMessageIDLen): true, + strings.Repeat("x", MaxMessageIDLen+1): false, + } { + if got := ValidMessageID(id); got != want { + t.Errorf("ValidMessageID(%q) = %v, want %v", id, got, want) + } + } +} + +func TestTypeName_Tagged(t *testing.T) { + t.Parallel() + if got := TypeName(TypeTagged); got != "TAGGED" { + t.Fatalf("TypeName(TypeTagged) = %q", got) + } +} + +func FuzzTaggedFrameRoundTrip(f *testing.F) { + f.Add(uint32(TypeText), []byte("hello"), "id-1", "") + f.Add(uint32(TypeJSON), []byte(`{}`), "", "reply") + f.Add(uint32(TypeBinary), []byte{0, 1, 2}, "a", "b") + f.Fuzz(func(t *testing.T, ftype uint32, payload []byte, messageID, replyTo string) { + if ftype == TypeTagged || ftype == TypeFile { + return // not a caller-level frame type / filename covered elsewhere + } + frame := &Frame{Type: ftype, Payload: payload, MessageID: messageID, ReplyTo: replyTo} + var buf bytes.Buffer + if err := WriteFrame(&buf, frame); err != nil { + if (messageID == "" || ValidMessageID(messageID)) && (replyTo == "" || ValidMessageID(replyTo)) { + t.Fatalf("WriteFrame rejected valid metadata %q/%q: %v", messageID, replyTo, err) + } + return + } + got, err := ReadFrame(&buf) + if err != nil { + t.Fatalf("ReadFrame: %v", err) + } + if got.Type != ftype || !bytes.Equal(got.Payload, payload) || got.MessageID != messageID || got.ReplyTo != replyTo { + t.Fatalf("round trip mismatch: %+v", got) + } + }) +} + +func FuzzTaggedFrameRead(f *testing.F) { + f.Add(rawTaggedPayload(`{"message_id":"a"}`, TypeText, []byte("x"))) + f.Add([]byte{0, 0}) + f.Add(bytes.Repeat([]byte{0xFF}, 64)) + f.Fuzz(func(t *testing.T, payload []byte) { + // Must not panic; a nil error must come with a non-tagged frame + // whose metadata is valid. + got, err := ReadFrame(bytes.NewReader(rawFrame(TypeTagged, payload))) + if err != nil { + return + } + if got.Type == TypeTagged { + t.Fatal("ReadFrame returned a nested TypeTagged frame") + } + if (got.MessageID != "" && !ValidMessageID(got.MessageID)) || (got.ReplyTo != "" && !ValidMessageID(got.ReplyTo)) { + t.Fatalf("ReadFrame accepted invalid metadata %q/%q", got.MessageID, got.ReplyTo) + } + }) +} + +// TestClient_Send_TaggedThroughDriver drives Client.Send over the real +// driver: the tagged frame goes out as TypeTagged and the ACK comes back. +func TestClient_Send_TaggedThroughDriver(t *testing.T) { + t.Parallel() + drv, d, c := dialClient(t) + t.Cleanup(func() { _ = drv.Close(); d.close() }) + + type outcome struct { + res *SendResult + err error + } + done := make(chan outcome, 1) + go func() { + res, err := c.Send(&Frame{Type: TypeText, Payload: []byte("q"), MessageID: "abc", ReplyTo: "xyz"}) + done <- outcome{res, err} + }() + + body := waitForCompleteFrame(t, d) + if got := binary.BigEndian.Uint32(body[0:4]); got != TypeTagged { + t.Fatalf("on-wire type = %d, want TypeTagged", got) + } + sent, err := ReadFrame(newByteReader(body)) + if err != nil { + t.Fatalf("decode sent frame: %v", err) + } + if sent.Type != TypeText || string(sent.Payload) != "q" || sent.MessageID != "abc" || sent.ReplyTo != "xyz" { + t.Fatalf("sent frame = %+v", sent) + } + + ack := frameBytes(t, &Frame{Type: TypeText, Payload: []byte("ACK TEXT 1 bytes")}) + push := make([]byte, 1+4+len(ack)) + push[0] = wireCmdRecv + binary.BigEndian.PutUint32(push[1:5], 0x42) + copy(push[5:], ack) + d.push(t, push) + + select { + case got := <-done: + if got.err != nil { + t.Fatalf("Send: %v", got.err) + } + if !got.res.Tagged || got.res.Duplicate || string(got.res.Ack.Payload) != "ACK TEXT 1 bytes" { + t.Fatalf("result = %+v", got.res) + } + case <-time.After(3 * time.Second): + t.Fatal("Send did not return after the ACK") + } + _ = c.Close() +} + +// TestMaxTaggedOverhead: the documented worst-case header overhead is what +// two maximal IDs actually cost on the wire. +func TestMaxTaggedOverhead(t *testing.T) { + t.Parallel() + id := strings.Repeat("x", MaxMessageIDLen) + payload, err := taggedWirePayload(&Frame{Type: TypeText, MessageID: id, ReplyTo: id}) + if err != nil { + t.Fatal(err) + } + if len(payload) != MaxTaggedOverhead || MaxTaggedOverhead != 293 { + t.Fatalf("maximal tagged overhead = %d bytes, MaxTaggedOverhead = %d, want both 293", len(payload), MaxTaggedOverhead) + } +} + +// countingWriter counts bytes written and discards them. +type countingWriter struct{ n int } + +func (w *countingWriter) Write(p []byte) (int, error) { w.n += len(p); return len(p), nil } + +// TestWriteFrame_TaggedFrameOverMaxFrameSize is part of review finding +// DX-REV-3: the tagged wrapper must not push a frame past MaxFrameSize, +// where the receiver would drop the connection. WriteFrame refuses such a +// frame before writing anything, and accepts one that fits exactly. +func TestWriteFrame_TaggedFrameOverMaxFrameSize(t *testing.T) { + t.Parallel() + overhead, err := taggedWirePayload(&Frame{Type: TypeBinary, MessageID: "m"}) + if err != nil { + t.Fatal(err) + } + // Untouched zero pages: this does not commit MaxFrameSize bytes of RAM + // until a write copies them. + big := make([]byte, MaxFrameSize) + + for _, tc := range []struct { + name string + n int + }{ + {"untagged size at the cap", int(MaxFrameSize)}, + {"one byte over once tagged", int(MaxFrameSize) - len(overhead) + 1}, + } { + var w countingWriter + err := WriteFrame(&w, &Frame{Type: TypeBinary, Payload: big[:tc.n], MessageID: "m"}) + if !errors.Is(err, ErrTaggedFrameTooLarge) { + t.Fatalf("%s: err = %v, want ErrTaggedFrameTooLarge", tc.name, err) + } + if w.n != 0 { + t.Fatalf("%s: %d bytes written before the error, want 0", tc.name, w.n) + } + } + + var w countingWriter + if err := WriteFrame(&w, &Frame{Type: TypeBinary, Payload: big[:int(MaxFrameSize)-len(overhead)], MessageID: "m"}); err != nil { + t.Fatalf("tagged frame that fits exactly: %v", err) + } + if w.n != 8+int(MaxFrameSize) { + t.Fatalf("exact fit wrote %d bytes, want %d", w.n, 8+int(MaxFrameSize)) + } +} + +// TestSend_TaggedFrameTooLargeSentUntagged: Client.Send delivers a frame +// that fits only untagged, untagged, and reports Tagged == false. +func TestSend_TaggedFrameTooLargeSentUntagged(t *testing.T) { + t.Parallel() + c2sR, c2sW := io.Pipe() + s2cR, s2cW := io.Pipe() + t.Cleanup(func() { _ = c2sW.Close(); _ = s2cR.Close() }) + type seen struct { + ftype uint32 + length uint32 + } + frames := make(chan seen, 4) + go func() { + // A receiver that checks the outer length like ReadFrame does but + // discards the payload, so the test does not buffer 64 MiB. + defer s2cW.Close() + for { + var hdr [8]byte + if _, err := io.ReadFull(c2sR, hdr[:]); err != nil { + return + } + ftype, length := binary.BigEndian.Uint32(hdr[0:4]), binary.BigEndian.Uint32(hdr[4:8]) + frames <- seen{ftype, length} + if length > MaxFrameSize { + // "frame too large": the receiver drops the connection, + // which fails the sender's pending write. + _ = c2sR.CloseWithError(errors.New("receiver dropped the connection: frame too large")) + return + } + if _, err := io.CopyN(io.Discard, c2sR, int64(length)); err != nil { + return + } + ack := fmt.Sprintf("ACK %s %d bytes", TypeName(ftype), length) + if err := WriteFrame(s2cW, &Frame{Type: TypeText, Payload: []byte(ack)}); err != nil { + return + } + } + }() + + payload := make([]byte, MaxFrameSize) + conn := struct { + io.Reader + io.Writer + }{s2cR, c2sW} + res, err := sendAndAwaitAck(conn, &Frame{Type: TypeBinary, Payload: payload, MessageID: NewMessageID()}) + if err != nil { + t.Fatalf("send: %v", err) + } + want := fmt.Sprintf("ACK BINARY %d bytes", MaxFrameSize) + if res.Tagged || string(res.Ack.Payload) != want { + t.Fatalf("result = %+v ack = %q, want untagged %q", res, res.Ack.Payload, want) + } + if got := <-frames; got.ftype != TypeBinary || got.length != MaxFrameSize { + t.Fatalf("receiver saw type %d length %d, want one untagged BINARY frame of %d bytes", got.ftype, got.length, MaxFrameSize) + } + select { + case extra := <-frames: + t.Fatalf("receiver saw a second frame %+v", extra) + default: + } +}