From fb04d9a6b4006813050c3397bae13c71ac6e487a Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 09:42:50 +0100 Subject: [PATCH 1/5] anomaly: late-delivery flag in the export (opt-in, #538) Add an opt-in export flag that marks the events of an observer burst delivering transmissions other observers reported minutes earlier: a reconnect queue stamped at delivery. It sets the existing ingest_backlog boolean, so the rate rule's exclude_ingest_backlog guard reads it unchanged; the library, replay and goldens are untouched. A row is already seen when another observer heard its transmission at least -late-gap before it and none within the gap. An observer's bin is a late burst when its window (the bin and both neighbours) holds at least -late-min-rows rows, at least -late-share of them already seen, with those delays spread over at least the gap. Unlike the #473 flag it needs no silence, so a real storm that an often-quiet observer hears first is not flagged. Four explicit flags, no defaults, exclusive with -backlog-*. Only the boolean leaves the export; the output is byte-identical under observer renaming. Relates to #538 Co-Authored-By: Claude Opus 5.5 --- cmd/anomaly/README.md | 55 ++- cmd/anomaly/backlog_473_test.go | 8 +- cmd/anomaly/export.go | 59 ++- cmd/anomaly/late.go | 227 ++++++++++ cmd/anomaly/late_538_test.go | 645 +++++++++++++++++++++++++++++ cmd/anomaly/late_sweep_538_test.go | 222 ++++++++++ cmd/anomaly/main.go | 4 +- 7 files changed, 1197 insertions(+), 23 deletions(-) create mode 100644 cmd/anomaly/late.go create mode 100644 cmd/anomaly/late_538_test.go create mode 100644 cmd/anomaly/late_sweep_538_test.go diff --git a/cmd/anomaly/README.md b/cmd/anomaly/README.md index c41874b1d..21c19ec1a 100644 --- a/cmd/anomaly/README.md +++ b/cmd/anomaly/README.md @@ -32,9 +32,10 @@ corescope-anomaly export -db /snap/meshcore.db -immutable \ names, public keys, observer columns, SNR/RSSI or resolved paths. A line carries payload type, the 1-byte channel hash (group payloads), route class, first hop and full route (hop prefixes as on the wire), frame and payload - size. The only exception is the optional - [ingest-backlog flag](#ingest-backlog-flag-optional-473). It reads - `observations.observer_idx` inside the export and writes a boolean only. + size. The only exceptions are the optional + [ingest-backlog flag](#ingest-backlog-flag-optional-473) and + [late-delivery flag](#late-delivery-flag-optional-538). They read + `observations.observer_idx` inside the export and write a boolean only. - **Opaque IDs.** A line's `id` is the export's own number: `tx1`, `tx2`, … in the order each transmission's first row is read. It is not the content hash, which would join the line to every CoreScope instance that heard the packet @@ -89,6 +90,49 @@ observer would look reconnected. A lookback that starts earlier is refused. The summary adds `ingest_backlog` with the scan's row counts and the number of flagged events. +### Late-delivery flag (optional, #538) + +```sh +corescope-anomaly export ... -late-window 1m -late-min-rows 10 -late-share 0.5 -late-gap 2m +``` + +The four flags go together (window and gap whole seconds in (0, 1h], min rows +in 1..1048576, share in (0, 1]); there are no defaults, and without them the +output is unchanged. They cannot be combined with the `-backlog-*` flags. They +set the same `"ingest_backlog": true`, so a rate rule's +`exclude_ingest_backlog` reads them unchanged. The flag marks the events of an +observer burst that delivers transmissions the rest of the network reported +minutes earlier: a queue sent on reconnect and stamped at delivery. + +- **Already seen.** For a row of observer O at time `t`, `e` is the latest + observation of the same transmission by another observer before + `t + gap`. The row is already seen when there is one and its delay + `d = t - e` is at least `-late-gap`: others heard it at least the gap + before, and none of them within the gap. +- **Late burst.** Bins are epoch-aligned and `-late-window` wide. O's bin k is + a late burst when O's rows in bins k-1, k and k+1 number at least + `-late-min-rows`, at least `-late-share` of them are already seen, and the + delays of the already-seen rows span at least `-late-gap`. The spread keeps + an observer whose clock runs behind by a constant offset from looking late. +- **Flag.** An event is flagged when its first observation (the one that sets + its `time`) is a row of O in a late burst of O. Rows with a NULL + `observer_idx` are never flagged. + +Unlike the #473 flag, it needs no silence: a real storm that an often-quiet +observer hears first is not flagged, because the other observers hear the +same transmissions within seconds or not at all. + +**Key-free.** As for the #473 flag: `observer_idx` stays in memory, the flag +is the only change to a line, and the output is byte-identical when observers +are renamed. + +**Cost.** One more index range scan on `observations.timestamp` over +`[start - settle - 2 window, end + settle + 2 window)`, under the same +`-max-rows` cap, with one index search per row over its transmission's +observations. Memory holds one entry per (observer, bin) with rows, capped. +The summary adds `late_delivery` with the scan's row counts, the rows already +seen and the number of flagged events. + Events are built by `anomaly.Normalizer`. An observation is route evidence when it lies at most `-settle` after the transmission's first observation. Rows without `observations.raw_hex` (written before #881) take their header and @@ -159,8 +203,9 @@ rules accept two optional fields: hex (`["81"]`). Leaving it out excludes nothing. See [Excluded channels](#excluded-channels). -Rate rules accept the optional `exclude_ingest_backlog` (#473). With `true`, -the rule ignores lines with `"ingest_backlog": true`: they are not counted in +Rate rules accept the optional `exclude_ingest_backlog` (#473; the #538 flag +sets the same field). With `true`, the rule ignores lines with +`"ingest_backlog": true`: they are not counted in its window, never trigger it and do not keep its episodes open. Leaving it out means `false`, and then the flag changes nothing. diff --git a/cmd/anomaly/backlog_473_test.go b/cmd/anomaly/backlog_473_test.go index 14a189109..36843ee65 100644 --- a/cmd/anomaly/backlog_473_test.go +++ b/cmd/anomaly/backlog_473_test.go @@ -104,11 +104,15 @@ func backlogScenario() []blTx { return txs } -func buildBacklogFixture(t *testing.T) *fixtureDB { +func buildBacklogFixture(t *testing.T) *fixtureDB { return buildFixtureOf(t, backlogScenario()) } + +// buildFixtureOf inserts each transmission as a one-hop flood GRP_TXT or +// ADVERT with its observations. +func buildFixtureOf(t *testing.T, txs []blTx) *fixtureDB { f := newFixtureDB(t, false) grp := grpPayload(0x0a, 20) adv := make([]byte, 100) - for i, x := range backlogScenario() { + for i, x := range txs { payload := grp if x.pt == 4 { payload = adv diff --git a/cmd/anomaly/export.go b/cmd/anomaly/export.go index 2df730fd2..c4c0eb064 100644 --- a/cmd/anomaly/export.go +++ b/cmd/anomaly/export.go @@ -28,7 +28,7 @@ import ( // time and the row IDs that join it to its transmission; never the content // hash, decoded_json, channel names, public keys, observer columns, // SNR/RSSI or resolved paths (except observer_idx for the opt-in backlog -// flag, which only a boolean leaves). A line's id is the export's own +// and late-delivery flags, which only a boolean leaves). A line's id is the export's own // number (see txNumbering). Bounded: one index range scan on observations.timestamp // over [start - settle, end + settle) with a hard -max-rows cap; per row a // primary-key lookup and one index search; memory is bounded by the caps @@ -44,7 +44,9 @@ import ( // the ingest time where the ingestor falls back to it (resolveRxTime). // // With -backlog-silence and -backlog-span the export also flags events -// whose first observation follows a silence of its observer (backlog.go). +// whose first observation follows a silence of its observer (backlog.go); +// with the -late-* flags instead, events that an observer delivered in a +// burst of transmissions others heard minutes earlier (late.go). const ( maxExportSpan = 31 * 24 * time.Hour @@ -69,6 +71,11 @@ type exportOptions struct { // BacklogSilence and BacklogSpan enable the ingest-backlog flag // (backlog.go); both or neither, 0 is off. BacklogSilence, BacklogSpan time.Duration + // LateWindow, LateMinRows, LateShare and LateGap enable the + // late-delivery flag (late.go); all or none, 0 is off. + LateWindow, LateGap time.Duration + LateMinRows int + LateShare float64 } func (o exportOptions) validate() error { @@ -88,7 +95,10 @@ func (o exportOptions) validate() error { case o.MaxRows <= 0 || o.MaxRows > maxExportRows: return fmt.Errorf("-max-rows must be in 1..%d", maxExportRows) } - return o.validateBacklog() + if err := o.validateBacklog(); err != nil { + return err + } + return o.validateLate() } // exportSummary counts rows and events; it holds no values from the data. @@ -112,6 +122,8 @@ type exportSummary struct { Normalizer anomaly.NormalizerStats `json:"normalizer"` // IngestBacklog is set only when the backlog flag is enabled. IngestBacklog *backlogSummary `json:"ingest_backlog,omitempty"` + // LateDelivery is set only when the late-delivery flag is enabled. + LateDelivery *lateSummary `json:"late_delivery,omitempty"` } // openReadOnly opens path with SQLite's mode=ro: the connection cannot @@ -157,7 +169,7 @@ func openSQLite(path, params string) (*sql.DB, error) { // ?3 seconds after the first observation of their transmission, in arrival // order, at most ?4 rows. ownRaw selects observations.raw_hex, which older // databases lack; observer adds observations.observer_idx for the backlog -// flag. +// or late-delivery flag. func exportQuery(ownRaw, observer bool) string { raw, obs := "NULL", "" if ownRaw { @@ -206,15 +218,29 @@ func exportRecords(opt exportOptions, w io.Writer) (exportSummary, error) { if err != nil { return sum, err } - var backlog *backlogTracker - if opt.backlog() { + // flag sets Observation.IngestBacklog; flaggedEvents counts the events + var flag interface { + flagged(observer sql.NullInt64, ts int64) bool + } + var flaggedEvents *int + switch { + case opt.backlog(): sum.IngestBacklog = &backlogSummary{} - if backlog, err = scanBacklog(db, opt, sum.IngestBacklog); err != nil { + b, err := scanBacklog(db, opt, sum.IngestBacklog) + if err != nil { + return sum, err + } + flag, flaggedEvents = b, &sum.IngestBacklog.EventsFlagged + case opt.late(): + sum.LateDelivery = &lateSummary{} + l, err := scanLate(db, opt, sum.LateDelivery) + if err != nil { return sum, err } + flag, flaggedEvents = l, &sum.LateDelivery.EventsFlagged } ids := newTxNumbering(int64(opt.Settle/time.Second), exportMaxPending) - rows, err := db.Query(exportQuery(ownRaw, backlog != nil), opt.Start.Add(-opt.Settle).Unix(), opt.End.Add(opt.Settle).Unix(), + rows, err := db.Query(exportQuery(ownRaw, flag != nil), opt.Start.Add(-opt.Settle).Unix(), opt.End.Add(opt.Settle).Unix(), int64(opt.Settle/time.Second), opt.MaxRows+1) if err != nil { return sum, err @@ -243,7 +269,7 @@ func exportRecords(opt exportOptions, w io.Writer) (exportSummary, error) { } sum.EventsWritten++ if e.IngestBacklog { - sum.IngestBacklog.EventsFlagged++ + *flaggedEvents++ } } return nil @@ -254,7 +280,7 @@ func exportRecords(opt exportOptions, w io.Writer) (exportSummary, error) { } var r obsRow dest := []interface{}{&r.id, &r.ts, &r.pathJSON, &r.ownRaw, &r.tx, &r.txRaw} - if backlog != nil { + if flag != nil { dest = append(dest, &r.observer) } if err := rows.Scan(dest...); err != nil { @@ -275,8 +301,8 @@ func exportRecords(opt exportOptions, w io.Writer) (exportSummary, error) { if note.pathUnknown { sum.PathUnknown++ } - if backlog != nil { - o.IngestBacklog = backlog.flagged(r.observer, r.ts) + if flag != nil { + o.IngestBacklog = flag.flagged(r.observer, r.ts) } if err := emit(norm.Add(o).Events); err != nil { return sum, err @@ -297,7 +323,7 @@ type obsRow struct { id, ts, tx int64 pathJSON, ownRaw sql.NullString txRaw string - observer sql.NullInt64 // read only for the backlog flag + observer sql.NullInt64 // read only for the backlog or late-delivery flag } // txNumbering gives each transmission an id of the export's own: "tx" and @@ -505,6 +531,10 @@ func runExport(args []string, stderr io.Writer) int { out := fs.String("out", "", "output JSON Lines file; must not exist (required)") silence := fs.Duration("backlog-silence", 0, "with -backlog-span: an observation with no observation of its observer in this long before it is a reconnect (optional, no default)") spn := fs.Duration("backlog-span", 0, "with -backlog-silence: flag events first heard by an observer within this long after its reconnect (optional, no default)") + lateWindow := fs.Duration("late-window", 0, "with the other -late-* flags: bin width of an observer's late burst (optional, no default)") + lateMinRows := fs.Int("late-min-rows", 0, "with the other -late-* flags: rows a late burst holds at least (optional, no default)") + lateShare := fs.Float64("late-share", 0, "with the other -late-* flags: share of a late burst's rows that are already seen, (0, 1] (optional, no default)") + lateGap := fs.Duration("late-gap", 0, "with the other -late-* flags: a row is already seen when another observer heard its transmission this long before (optional, no default)") if err := fs.Parse(args); err != nil { return exitUsage } @@ -516,7 +546,8 @@ func runExport(args []string, stderr io.Writer) int { return usageErr(errors.New("unexpected arguments")) } opt := exportOptions{DB: *dbPath, Settle: *settle, MaxRows: *maxRows, Immutable: *immutable, - BacklogSilence: *silence, BacklogSpan: *spn} + BacklogSilence: *silence, BacklogSpan: *spn, + LateWindow: *lateWindow, LateMinRows: *lateMinRows, LateShare: *lateShare, LateGap: *lateGap} var err error if opt.Start, err = parseTime("start", *start); err != nil { return usageErr(err) diff --git a/cmd/anomaly/late.go b/cmd/anomaly/late.go new file mode 100644 index 000000000..93fa2dbc0 --- /dev/null +++ b/cmd/anomaly/late.go @@ -0,0 +1,227 @@ +package main + +import ( + "database/sql" + "errors" + "fmt" + "time" +) + +// Late-delivery flag (#538, opt-in with -late-window, -late-min-rows, +// -late-share and -late-gap; no defaults). It writes the same +// "ingest_backlog": true as the #473 flag, so a rate rule's +// exclude_ingest_backlog guard reads it unchanged; the two flags exclude +// each other. +// +// It marks the events of an observer burst that delivers transmissions +// other observers heard minutes earlier: queued packets stamped when the +// observer's uploader finally sent them, not when they were heard. The #473 +// flag (first observations after a silence) also marked a real storm that +// an often-quiet observer heard first; this one needs the burst to repeat +// what the rest of the network already reported. +// +// - Delay. For a row r of observer O at time t, e is the latest +// observation of the same transmission by another (non-NULL) observer +// before t + gap, over the whole database. r is already seen when there +// is one and d = t - e >= gap: other observers heard the transmission +// at least gap before r and none of them within gap of r. A live row +// has a live peer within seconds; one observer whose clock runs early +// does not make the others look late. +// - Window. Bins are epoch-aligned, -late-window wide. The window of +// (O, bin k) is O's rows in bins k-1, k and k+1. +// - Burst. (O, k) is a late burst when its window holds at least +// -late-min-rows rows, at least -late-share of them are already seen, +// and the delays of those span at least gap (max d - min d). The spread +// separates a queue, which holds transmissions heard over the whole +// outage, from an observer whose clock runs behind by a constant +// offset: every one of its delays is about that offset. +// - Flag. An event is flagged when its first observation (the one that +// sets the event's time) is a row of O in a late burst (O, its bin). +// Rows with a NULL observer are never flagged. +// +// The observer is read as observations.observer_idx and used only inside +// the export, as a map key and for "another observer"; only the boolean +// leaves, so the output does not change when observers are renamed. +// +// Activity is one more index range scan on observations.timestamp over +// [start - settle - 2*window, end + settle + 2*window), which holds every +// window of a row the export reads, under the same -max-rows cap; per row +// one index search over the observations of its transmission. Memory is +// one entry per (observer, bin) with rows, capped. + +const ( + maxLateDuration = time.Hour + maxLateMinRows = 1 << 20 + // Capacity of the activity scan (not a detection threshold); exceeding + // it fails the export. + exportMaxLateBins = 1 << 22 +) + +// lateSummary counts the activity scan; it holds no values from the data. +type lateSummary struct { + // Rows is the observation rows of the activity scan. + Rows int `json:"rows"` + // RowsWithoutObserver counts rows whose observer_idx is NULL. + RowsWithoutObserver int `json:"rows_without_observer"` + // RowsAlreadySeen counts rows whose delay is at least -late-gap. + RowsAlreadySeen int `json:"rows_already_seen"` + // EventsFlagged counts written events with ingest_backlog set. + EventsFlagged int `json:"events_flagged"` +} + +func validLateDuration(d time.Duration) bool { + return d > 0 && d <= maxLateDuration && d%time.Second == 0 +} + +func (o exportOptions) late() bool { + return o.LateWindow != 0 || o.LateMinRows != 0 || o.LateShare != 0 || o.LateGap != 0 +} + +func (o exportOptions) validateLate() error { + switch { + case !o.late(): + return nil + case o.backlog(): + return errors.New("the -backlog-* and -late-* flags exclude each other") + case !validLateDuration(o.LateWindow) || !validLateDuration(o.LateGap) || + o.LateMinRows < 1 || o.LateMinRows > maxLateMinRows || !(o.LateShare > 0 && o.LateShare <= 1): + return fmt.Errorf("-late-window, -late-min-rows, -late-share and -late-gap go together: "+ + "whole seconds in (0, 1h], 1..%d rows and a share in (0, 1]", maxLateMinRows) + } + return nil +} + +// lateQuery selects the observer, time and elsewhere time e (NULL when +// there is none) of every observation in [?1, ?2), in time order, at most +// ?3 rows; ?4 is the gap in seconds. +const lateQuery = `SELECT o.observer_idx, o.timestamp, + (SELECT MAX(p.timestamp) FROM observations p + WHERE p.transmission_id = o.transmission_id AND p.timestamp < o.timestamp + ?4 + AND p.observer_idx IS NOT NULL AND p.observer_idx <> o.observer_idx) +FROM observations o +WHERE o.timestamp >= ?1 AND o.timestamp < ?2 +ORDER BY o.timestamp, o.id +LIMIT ?3` + +type lateKey struct{ observer, bin int64 } + +// lateBin counts one observer's rows in one bin; dmin and dmax are the +// least and greatest delay of its already-seen rows (when seen > 0). +type lateBin struct { + rows, seen int + dmin, dmax int64 +} + +func (b *lateBin) add(o lateBin) { + if o.seen > 0 { + if b.seen == 0 || o.dmin < b.dmin { + b.dmin = o.dmin + } + if b.seen == 0 || o.dmax > b.dmax { + b.dmax = o.dmax + } + } + b.rows += o.rows + b.seen += o.seen +} + +// lateTracker holds the bins of each observer, built by see and read by +// flagged. +type lateTracker struct { + window, gap int64 + minRows int + share float64 + maxBins int + bins map[lateKey]*lateBin +} + +func newLateTracker(opt exportOptions, maxBins int) *lateTracker { + return &lateTracker{window: int64(opt.LateWindow / time.Second), gap: int64(opt.LateGap / time.Second), + minRows: opt.LateMinRows, share: opt.LateShare, maxBins: maxBins, bins: make(map[lateKey]*lateBin)} +} + +func floorDiv(a, b int64) int64 { + q := a / b + if a%b != 0 && (a < 0) != (b < 0) { + q-- + } + return q +} + +// see records one row of observer at ts; elsewhere is the latest +// observation of its transmission by another observer before ts + gap. It +// reports whether the row is already seen. +func (l *lateTracker) see(observer, ts int64, elsewhere sql.NullInt64) (bool, error) { + k := lateKey{observer, floorDiv(ts, l.window)} + b := l.bins[k] + if b == nil { + if len(l.bins) >= l.maxBins { + return false, fmt.Errorf("more than %d (observer, bin) entries in the late-delivery activity scan", l.maxBins) + } + b = &lateBin{} + l.bins[k] = b + } + r := lateBin{rows: 1} + if d := ts - elsewhere.Int64; elsewhere.Valid && d >= l.gap { + r.seen, r.dmin, r.dmax = 1, d, d + } + b.add(r) + return r.seen == 1, nil +} + +// flagged reports whether an observation of observer at ts lies in a late +// burst of that observer. +func (l *lateTracker) flagged(observer sql.NullInt64, ts int64) bool { + if !observer.Valid { + return false + } + k := floorDiv(ts, l.window) + var w lateBin + for i := k - 1; i <= k+1; i++ { + if b := l.bins[lateKey{observer.Int64, i}]; b != nil { + w.add(*b) + } + } + return w.rows >= l.minRows && float64(w.seen) >= l.share*float64(w.rows) && + w.seen > 0 && w.dmax-w.dmin >= l.gap +} + +// scanLate reads observer activity for the export range. +func scanLate(db *sql.DB, opt exportOptions, sum *lateSummary) (*lateTracker, error) { + ok, err := hasColumn(db, "observations", "observer_idx") + if err != nil { + return nil, err + } + if !ok { + return nil, errors.New("the late-delivery flag needs observations.observer_idx") + } + l := newLateTracker(opt, exportMaxLateBins) + margin := opt.Settle + 2*opt.LateWindow + rows, err := db.Query(lateQuery, opt.Start.Add(-margin).Unix(), opt.End.Add(margin).Unix(), opt.MaxRows+1, l.gap) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + if sum.Rows++; sum.Rows > opt.MaxRows { + return nil, fmt.Errorf("more than -max-rows %d observation rows in the late-delivery activity scan", opt.MaxRows) + } + var observer, elsewhere sql.NullInt64 + var ts int64 + if err := rows.Scan(&observer, &ts, &elsewhere); err != nil { + return nil, err + } + if !observer.Valid { + sum.RowsWithoutObserver++ + continue + } + seen, err := l.see(observer.Int64, ts, elsewhere) + if err != nil { + return nil, err + } + if seen { + sum.RowsAlreadySeen++ + } + } + return l, rows.Err() +} diff --git a/cmd/anomaly/late_538_test.go b/cmd/anomaly/late_538_test.go new file mode 100644 index 000000000..6b883cd61 --- /dev/null +++ b/cmd/anomaly/late_538_test.go @@ -0,0 +1,645 @@ +package main + +import ( + "bufio" + "bytes" + "database/sql" + "encoding/json" + "fmt" + "io" + "math" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + "testing" + "time" + + "github.com/meshcore-analyzer/anomaly" +) + +// Issue #538: the export's late-delivery flag. An event is flagged when its +// first observation lies in a burst of its observer that repeats +// transmissions other observers heard minutes earlier. Only the existing +// ingest_backlog boolean is written. + +const ( + ltWindow = time.Minute + ltMinRows = 10 + ltShare = 0.5 + ltGap = 2 * time.Minute +) + +// Observers of the late-delivery fixture (observations.observer_idx). +const ( + ltA = 81001 // live, hears every background ADVERT first + ltB = 81002 // live, one second after A + ltDrain = 81003 // live, then a 40-minute outage, then delivers its queue at once + ltQuiet = 81004 // silent, then first to hear a real storm (the #473 loss) + ltSkew = 81005 // clock 167 s behind: its rows look early, everyone else's late + ltIsolated = 81006 // silent, then a real storm nobody else hears +) + +// ltGroup names the part of the scenario a transmission belongs to. +type ltGroup string + +const ( + ltBackground ltGroup = "background" + ltDrainQueue ltGroup = "drain backlog" + ltQuietStorm ltGroup = "storm, quiet observer first" + ltSkewStorm ltGroup = "storm, skewed observer" + ltIsoStorm ltGroup = "storm, isolated observer" + ltNoObserver ltGroup = "no observer" +) + +type ltTx struct { + blTx + group ltGroup +} + +func ltMin(m int, s int) time.Duration { + return time.Duration(m)*time.Minute + time.Duration(s)*time.Second +} + +// lateScenario lists the fixture; flag marks the events the late-delivery +// flag must set. +func lateScenario() []ltTx { + var txs []ltTx + add := func(g ltGroup, at time.Time, pt int, flag bool, obs ...blObs) { + txs = append(txs, ltTx{blTx{at: at, pt: pt, obs: obs, flag: flag}, g}) + } + // background: one ADVERT a minute at m:37, heard by A, then B, then + // Drain; Drain is out for [5m, 45m) and delivers those rows from 45m00s + // on, one a second, stamped with its delivery time; Skew hears the ADVERTs + // of minutes 60..85, stamped 167 s early + for m := -150; m < 130; m++ { + at := blAt(ltMin(m, 37)) + obs := []blObs{{ltA, 0}, {ltB, time.Second}} + switch { + case m >= 5 && m < 45: + obs = append(obs, blObs{ltDrain, blAt(ltMin(45, m-5)).Sub(at)}) + case m != 45: + obs = append(obs, blObs{ltDrain, 2 * time.Second}) + } + if m >= 60 && m <= 85 { + obs = append(obs, blObs{ltSkew, -167 * time.Second}) + } + add(ltBackground, at, 4, false, obs...) + } + // the queue also holds 25 GRP_TXT nobody else heard: the backlog storm + for i := 0; i < 25; i++ { + add(ltDrainQueue, blAt(ltMin(45, i)), 5, true, blObs{ltDrain, 0}) + } + // a first observation without an observer inside Drain's burst, and + // one with Drain second: neither is flagged + add(ltNoObserver, blAt(ltMin(45, 45)), 4, false, blObs{0, 0}, blObs{ltDrain, time.Second}) + // a real storm of 30 GRP_TXT first heard by an observer that was silent + // for hours, then by A; the #473 flag marks it + for i := 0; i < 30; i++ { + add(ltQuietStorm, blAt(ltMin(55, i)), 5, false, blObs{ltQuiet, 0}, blObs{ltA, 2 * time.Second}) + } + // a real storm of 40 GRP_TXT heard by A; Skew hears three in four, so + // A's rows around it are mostly "already seen", but all with the same + // delay (Skew's clock offset) + for i := 0; i < 40; i++ { + obs := []blObs{{ltA, 0}} + if i%4 != 3 { + obs = append(obs, blObs{ltSkew, -161 * time.Second}) + } + add(ltSkewStorm, blAt(ltMin(75, i)), 5, false, obs...) + } + // a real storm of 25 GRP_TXT that one silent observer hears alone + for i := 0; i < 25; i++ { + add(ltIsoStorm, blAt(ltMin(95, i)), 5, false, blObs{ltIsolated, 0}) + } + return txs +} + +// eventTime is the time of a transmission's event: its first observation. +func (x ltTx) eventTime() time.Time { + t := x.at.Add(x.obs[0].d) + for _, o := range x.obs[1:] { + if at := x.at.Add(o.d); at.Before(t) { + t = at + } + } + return t +} + +func buildLateFixture(t *testing.T) *fixtureDB { + var txs []blTx + for _, x := range lateScenario() { + txs = append(txs, x.blTx) + } + return buildFixtureOf(t, txs) +} + +func ltOptions(db string) exportOptions { + opt := fxOptions(db) + opt.LateWindow, opt.LateMinRows, opt.LateShare, opt.LateGap = ltWindow, ltMinRows, ltShare, ltGap + return opt +} + +// ltWant maps each in-range event time to its scenario transmission. +func ltWant(t *testing.T) map[time.Time]ltTx { + t.Helper() + want := map[time.Time]ltTx{} + for _, x := range lateScenario() { + at := x.eventTime() + if at.Before(fxT) || !at.Before(fxAt(2*time.Hour)) { + continue + } + if _, dup := want[at]; dup { + t.Fatalf("scenario: two events at %v", at.Sub(fxT)) + } + want[at] = x + } + return want +} + +// flagsByGroup counts events and flagged events per group. +func flagsByGroup(t *testing.T, out []byte, want map[time.Time]ltTx) map[ltGroup][2]int { + t.Helper() + rs := readRecords(t, out) + if len(rs) != len(want) { + t.Fatalf("%d lines, want %d", len(rs), len(want)) + } + g := map[ltGroup][2]int{} + for _, r := range rs { + x, ok := want[r.Time] + if !ok { + t.Fatalf("unexpected line at %v", r.Time.Sub(fxT)) + } + c := g[x.group] + c[0]++ + if r.IngestBacklog { + c[1]++ + } + g[x.group] = c + } + return g +} + +// The flag follows the definition: exactly the backlog storm Drain delivers +// is flagged, and none of the three real storms, while the #473 flag +// (first observations after a 10-minute silence, for 2 h) marks the +// backlog and also every event of the storms a quiet, skewed or isolated +// observer heard first. +func TestExportLateDeliveryFlagsTheDrainBacklogOnly(t *testing.T) { + f := buildLateFixture(t) + f.tx(fxHash(9999), frameHex(1, 4, 1, nil, make([]byte, 100)), 4, fxObs{at: fxAt(-3 * time.Hour), observer: ltA, + raw: frameHex(1, 4, 1, nil, make([]byte, 100))}) // lookback data for the #473 flag + want := ltWant(t) + got, sum := runExportTo(t, ltOptions(f.path)) + var wrong []string + for _, r := range readRecords(t, got) { + if x := want[r.Time]; r.IngestBacklog != x.flag { + wrong = append(wrong, fmt.Sprintf("%v (%s): flag %v, want %v", r.Time.Sub(fxT), x.group, r.IngestBacklog, x.flag)) + } + } + if len(wrong) > 0 { + t.Fatalf("%d events flagged wrongly:\n%s", len(wrong), strings.Join(wrong, "\n")) + } + if sum.LateDelivery == nil || sum.LateDelivery.EventsFlagged != 25 || sum.IngestBacklog != nil { + t.Fatalf("summary %+v / %+v, want 25 flagged events under late_delivery only", sum.LateDelivery, sum.IngestBacklog) + } + late := flagsByGroup(t, got, want) + + old := fxOptions(f.path) + old.BacklogSilence, old.BacklogSpan = 10*time.Minute, 2*time.Hour + oldOut, _ := runExportTo(t, old) + prev := flagsByGroup(t, oldOut, want) + groups := []ltGroup{ltBackground, ltDrainQueue, ltQuietStorm, ltSkewStorm, ltIsoStorm, ltNoObserver} + for _, g := range groups { + t.Logf("%-30s events %3d #473 flag %3d late-delivery flag %3d", g, late[g][0], prev[g][1], late[g][1]) + } + for g, w := range map[ltGroup]int{ltDrainQueue: 25, ltQuietStorm: 30, ltSkewStorm: 30, ltIsoStorm: 25} { + if prev[g][1] != w { + t.Fatalf("#473 flag on %s: %d flagged, want %d", g, prev[g][1], w) + } + } +} + +// The late-delivery summary counts the activity scan: its rows cover +// [start - settle - 2 window, end + settle + 2 window), rows without an +// observer, and rows whose transmission another observer heard at least the +// gap earlier and none within the gap. +func TestExportLateDeliverySummaryCountsTheScan(t *testing.T) { + f := buildLateFixture(t) + _, sum := runExportTo(t, ltOptions(f.path)) + lo := fxT.Add(-fxSettle - 2*ltWindow).Unix() + hi := fxAt(2*time.Hour + fxSettle + 2*ltWindow).Unix() + var rows, null, seen int + if err := f.db.QueryRow(`SELECT COUNT(*), COUNT(*) - COUNT(observer_idx), + SUM(EXISTS (SELECT 1 FROM observations p WHERE p.transmission_id = o.transmission_id + AND p.observer_idx IS NOT NULL AND p.observer_idx <> o.observer_idx AND p.timestamp <= o.timestamp - ?1) + AND NOT EXISTS (SELECT 1 FROM observations p WHERE p.transmission_id = o.transmission_id + AND p.observer_idx IS NOT NULL AND p.observer_idx <> o.observer_idx + AND p.timestamp > o.timestamp - ?1 AND p.timestamp < o.timestamp + ?1)) + FROM observations o WHERE timestamp >= ?2 AND timestamp < ?3`, int64(ltGap/time.Second), lo, hi).Scan(&rows, &null, &seen); err != nil { + t.Fatal(err) + } + if l := sum.LateDelivery; l == nil || l.Rows != rows || l.RowsWithoutObserver != null || l.RowsAlreadySeen != seen || + null != 1 || seen == 0 { + t.Fatalf("summary %+v, want rows %d, without observer %d, already seen %d", sum.LateDelivery, rows, null, seen) + } +} + +// Key-free: the flag is the only difference to the export without it, the +// only new key is "ingest_backlog" with the value true, and no sentinel or +// observer index reaches the output. +func TestExportLateDeliveryAddsOnlyTheBoolean(t *testing.T) { + f := buildLateFixture(t) + plain, _ := runExportTo(t, fxOptions(f.path)) + got, sum := runExportTo(t, ltOptions(f.path)) + if !bytes.Equal(stripFlag(got), plain) { + t.Fatal("the flagged export differs from the plain one in more than the flag") + } + if n := bytes.Count(got, []byte(`"ingest_backlog"`)); n == 0 || n != sum.LateDelivery.EventsFlagged || + n != bytes.Count(got, []byte(`,"ingest_backlog":true}`+"\n")) { + t.Fatalf("%d flags, summary %+v", n, sum.LateDelivery) + } + for _, s := range []string{sentinelText, sentinelChannel, sentinelPubkey, sentinelObserver, sentinelObsID, + sentinelResolved, "SENTINEL", "-7.5", "-101", `"rx"`, "observer", "late", "8100"} { + if bytes.Contains(got, []byte(s)) { + t.Fatalf("export output contains %q", s) + } + } + keys := map[string]bool{} + sc := bufio.NewScanner(bytes.NewReader(got)) + for sc.Scan() { + var m map[string]json.RawMessage + if err := json.Unmarshal(sc.Bytes(), &m); err != nil { + t.Fatal(err) + } + for k, v := range m { + keys[k] = true + if k == "ingest_backlog" && string(v) != "true" { + t.Fatalf("ingest_backlog = %s", v) + } + } + } + var ks []string + for k := range keys { + ks = append(ks, k) + } + sort.Strings(ks) + if got, want := strings.Join(ks, ","), "channel,decryption,final_at,first_hop,full_route,id,ingest_backlog,"+ + "payload_bytes,payload_type,route_class,route_kind,time,traffic,wire_bytes"; got != want { + t.Fatalf("keys %s, want %s", got, want) + } +} + +// The output depends on observers only through which rows share one: +// renaming every observer (any bijection of observer_idx) gives a +// byte-identical export and summary. +func TestExportLateDeliveryIsInvariantUnderObserverRenaming(t *testing.T) { + f := buildLateFixture(t) + before, sumBefore := runExportTo(t, ltOptions(f.path)) + for _, q := range []string{ + `UPDATE observations SET observer_idx = 900000 - observer_idx WHERE observer_idx IS NOT NULL`, + `UPDATE observations SET observer_idx = CASE observer_idx WHEN 818999 THEN 7 WHEN 818997 THEN 3 ELSE observer_idx END`, + } { + if _, err := f.db.Exec(q); err != nil { + t.Fatal(err) + } + after, sumAfter := runExportTo(t, ltOptions(f.path)) + if !bytes.Equal(after, before) || *sumAfter.LateDelivery != *sumBefore.LateDelivery { + t.Fatalf("renaming observers changed the export (%s)", q) + } + } + var n int + f.db.QueryRow(`SELECT COUNT(*) FROM observations WHERE observer_idx IN (3, 7)`).Scan(&n) + if n == 0 { + t.Fatal("the renaming touched no rows") + } +} + +// Without the -late-* options the export is unchanged: no flag and no +// late_delivery summary. The four options come together, each in range, +// and never with the #473 options. +func TestExportLateDeliveryIsOptIn(t *testing.T) { + f := buildLateFixture(t) + got, sum := runExportTo(t, fxOptions(f.path)) + if bytes.Contains(got, []byte("ingest_backlog")) || sum.LateDelivery != nil { + t.Fatalf("flag or summary without the options: %+v", sum.LateDelivery) + } + if b, _ := json.Marshal(sum); bytes.Contains(b, []byte("late")) { + t.Fatalf("summary names the late-delivery flag: %s", b) + } + if err := ltOptions(f.path).validate(); err != nil { + t.Fatal(err) + } + for name, mod := range map[string]func(*exportOptions){ + "window only": func(o *exportOptions) { *o = fxOptions(o.DB); o.LateWindow = ltWindow }, + "min rows only": func(o *exportOptions) { *o = fxOptions(o.DB); o.LateMinRows = ltMinRows }, + "share only": func(o *exportOptions) { *o = fxOptions(o.DB); o.LateShare = ltShare }, + "gap only": func(o *exportOptions) { *o = fxOptions(o.DB); o.LateGap = ltGap }, + "no window": func(o *exportOptions) { o.LateWindow = 0 }, + "no gap": func(o *exportOptions) { o.LateGap = 0 }, + "no min rows": func(o *exportOptions) { o.LateMinRows = 0 }, + "no share": func(o *exportOptions) { o.LateShare = 0 }, + "negative window": func(o *exportOptions) { o.LateWindow = -time.Minute }, + "window 1h1s": func(o *exportOptions) { o.LateWindow = time.Hour + time.Second }, + "gap 1h1s": func(o *exportOptions) { o.LateGap = time.Hour + time.Second }, + "fractional gap": func(o *exportOptions) { o.LateGap = 1500 * time.Millisecond }, + "min rows 2^20+1": func(o *exportOptions) { o.LateMinRows = maxLateMinRows + 1 }, + "share 1.01": func(o *exportOptions) { o.LateShare = 1.01 }, + "negative share": func(o *exportOptions) { o.LateShare = -0.5 }, + "NaN share": func(o *exportOptions) { o.LateShare = math.NaN() }, + "with #473 flag": func(o *exportOptions) { o.BacklogSilence, o.BacklogSpan = blSilence, blSpan }, + } { + opt := ltOptions(f.path) + mod(&opt) + if err := opt.validate(); err == nil { + t.Fatalf("%s: accepted", name) + } + } + for _, ok := range []func(*exportOptions){ + func(o *exportOptions) { o.LateWindow, o.LateGap = time.Hour, time.Second }, + func(o *exportOptions) { o.LateMinRows, o.LateShare = maxLateMinRows, 1 }, + } { + opt := ltOptions(f.path) + ok(&opt) + if err := opt.validate(); err != nil { + t.Fatalf("a bound value was refused: %v", err) + } + } +} + +// The activity scan reads only the observer index, time, row id and +// transmission row id; it joins no table and touches no other column. +func TestLateQueryReadsOnlyActivity(t *testing.T) { + s := strings.NewReplacer("o.observer_idx", "", "p.observer_idx", "").Replace(strings.ToLower(lateQuery)) + for _, col := range []string{"decoded_json", "from_pubkey", "channel_hash", "observer", "snr", "rssi", "score", + "direction", "resolved_path", "name", "first_seen", "hash", "*", "iata", "join", "raw_hex", "path_json", "transmissions"} { + if strings.Contains(s, col) { + t.Fatalf("activity scan reads %q:\n%s", col, lateQuery) + } + } +} + +// The activity scan is one index search on observations.timestamp in time +// order, and its subquery one index search per row on the transmission's +// observations (no scan, no sort). +func TestLateQueryPlanIsBounded(t *testing.T) { + f := buildLateFixture(t) + db, err := openReadOnly(f.path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + rows, err := db.Query(`EXPLAIN QUERY PLAN `+lateQuery, 0, 1, 1, 1) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var p []string + for rows.Next() { + var id, parent, notused int + var detail string + if err := rows.Scan(&id, &parent, ¬used, &detail); err != nil { + t.Fatal(err) + } + p = append(p, detail) + } + plan := strings.Join(p, "\n") + for _, w := range []string{"SEARCH o USING INDEX idx_observations_timestamp (timestamp>? AND timestamp= min rows, already-seen +// share >= share, and already-seen delays spread over >= gap; a row is +// already seen at a delay of exactly the gap. The window is the bin and its +// two neighbours. +func TestLateTrackerBurst(t *testing.T) { + type row struct { + ts int64 + d int64 // < 0: no observation elsewhere + } + cases := []struct { + name string + rows []row + want bool + }{ + {"at every bound", []row{{600, 120}, {601, 240}, {602, -1}, {603, -1}}, true}, + {"one row short", []row{{600, 120}, {601, 240}, {602, -1}}, false}, + {"share just below", []row{{600, 120}, {601, 240}, {602, -1}, {603, -1}, {604, -1}}, false}, + {"delay one second below the gap", []row{{600, 119}, {601, 240}, {602, 240}, {603, -1}}, false}, + {"spread one second short", []row{{600, 121}, {601, 240}, {602, -1}, {603, -1}}, false}, + {"constant offset (skewed clock)", []row{{600, 300}, {601, 300}, {602, 300}, {603, 300}}, false}, + {"neighbour bins count", []row{{540, 120}, {601, 240}, {660, -1}, {719, -1}}, true}, + {"two bins away does not", []row{{540, 120}, {601, 240}, {660, -1}, {720, -1}}, false}, + } + for _, c := range cases { + l := ltTracker(4, 100) + for _, r := range c.rows { + e := sql.NullInt64{} + if r.d >= 0 { + e = nullInt(r.ts - r.d) + } + if _, err := l.see(1, r.ts, e); err != nil { + t.Fatal(err) + } + } + if got := l.flagged(nullInt(1), 601); got != c.want { + t.Fatalf("%s: flagged %v, want %v", c.name, got, c.want) + } + } + l := ltTracker(4, 100) + for _, r := range []row{{600, 120}, {601, 240}, {602, -1}, {603, -1}} { + e := sql.NullInt64{} + if r.d >= 0 { + e = nullInt(r.ts - r.d) + } + l.see(1, r.ts, e) + } + for _, c := range []struct { + ts int64 + want bool + }{{539, false}, {540, true}, {719, true}, {720, false}} { + if got := l.flagged(nullInt(1), c.ts); got != c.want { + t.Fatalf("t=%d: flagged %v, want %v", c.ts, got, c.want) + } + } + if l.flagged(nullInt(2), 600) || l.flagged(sql.NullInt64{}, 600) { + t.Fatal("another or a NULL observer was flagged") + } +} + +// Bins are epoch-aligned also before 1970, and see reports already-seen rows. +func TestLateTrackerBinsAndSeen(t *testing.T) { + for _, c := range []struct{ a, b, want int64 }{{-1, 60, -1}, {-60, 60, -1}, {-61, 60, -2}, {0, 60, 0}, {59, 60, 0}, {60, 60, 1}} { + if got := floorDiv(c.a, c.b); got != c.want { + t.Fatalf("floorDiv(%d, %d) = %d, want %d", c.a, c.b, got, c.want) + } + } + l := ltTracker(1, 100) + if seen, _ := l.see(1, 1000, nullInt(880)); !seen { + t.Fatal("a delay of the gap is not already seen") + } + if seen, _ := l.see(1, 1000, nullInt(881)); seen { + t.Fatal("a delay below the gap is already seen") + } + if seen, _ := l.see(1, 1000, sql.NullInt64{}); seen { + t.Fatal("a row nobody else heard is already seen") + } +} + +// The tracker holds at most maxBins (observer, bin) entries; one more +// fails the export instead of growing. +func TestLateTrackerIsBounded(t *testing.T) { + l := ltTracker(1, 2) + for _, r := range []struct{ o, ts int64 }{{1, 0}, {1, 59}, {2, 0}, {2, 30}} { + if _, err := l.see(r.o, r.ts, sql.NullInt64{}); err != nil { + t.Fatal(err) + } + } + if _, err := l.see(1, 60, sql.NullInt64{}); err == nil { + t.Fatal("a third entry was accepted") + } +} + +func ltArgs(db, out string, extra ...string) []string { + return append(exportArgs(db, out, "-late-window", "1m", "-late-min-rows", "10", "-late-share", "0.5", "-late-gap", "2m"), extra...) +} + +// The command takes the four options as explicit flags and prints the +// late-delivery counts in its summary. +func TestExportCommandLateDeliveryFlags(t *testing.T) { + f := buildLateFixture(t) + dir := t.TempDir() + out := filepath.Join(dir, "events.jsonl") + var stdout, stderr bytes.Buffer + if code := run(ltArgs(f.path, out), nil, &stdout, &stderr); code != 0 { + t.Fatalf("exit %d: %s", code, stderr.String()) + } + want, _ := runExportTo(t, ltOptions(f.path)) + if got, err := os.ReadFile(out); err != nil || !bytes.Equal(got, want) { + t.Fatalf("output differs from exportRecords (%v)", err) + } + if !regexp.MustCompile(`"late_delivery": \{\s*"rows": \d+,\s*"rows_without_observer": 1,\s*"rows_already_seen": \d+,\s*"events_flagged": 25\s*\}`). + Match(stderr.Bytes()) { + t.Fatalf("summary shape:\n%s", stderr.String()) + } + for name, args := range map[string][]string{ + "window only": exportArgs(f.path, filepath.Join(dir, "a"), "-late-window", "1m"), + "no share": exportArgs(f.path, filepath.Join(dir, "b"), "-late-window", "1m", "-late-min-rows", "10", "-late-gap", "2m"), + "share 2": exportArgs(f.path, filepath.Join(dir, "c"), "-late-window", "1m", "-late-min-rows", "10", "-late-share", "2", "-late-gap", "2m"), + "with #473": ltArgs(f.path, filepath.Join(dir, "d"), "-backlog-silence", "10m", "-backlog-span", "30m"), + } { + var so, se bytes.Buffer + if code := run(args, nil, &so, &se); code != 2 { + t.Fatalf("%s: exit %d, want 2 (%s)", name, code, se.String()) + } + } +} + +// End to end: the unguarded rate rule fires on the backlog storm and the +// three real storms. Guarded by the late-delivery flag it loses only the +// backlog storm; guarded by the #473 flag it loses all four, because a +// quiet, a skewed or an isolated observer heard each real storm first. +func TestExportThenGuardedReplayIgnoresTheLateBacklog(t *testing.T) { + f := buildLateFixture(t) + f.tx(fxHash(9999), frameHex(1, 4, 1, nil, make([]byte, 100)), 4, fxObs{at: fxAt(-3 * time.Hour), observer: ltA, + raw: frameHex(1, 4, 1, nil, make([]byte, 100))}) + late, _ := runExportTo(t, ltOptions(f.path)) + oldOpt := fxOptions(f.path) + oldOpt.BacklogSilence, oldOpt.BacklogSpan = 10*time.Minute, 2*time.Hour + old, _ := runExportTo(t, oldOpt) + activations := func(in []byte, guard bool) []time.Duration { + g := "" + if guard { + g = `, "exclude_ingest_backlog": true` + } + cfg, err := anomaly.ParseReplayConfig([]byte(`{ + "limits": {"max_keys_per_scope": 1000, "state_ttl": "48h", "dedup_horizon": "6h", "max_dedup_ids": 100000, + "reorder_delay": "0s", "max_reorder_buffer": 64, "max_window_entries": 4096, + "max_candidates_per_call": 1000, "max_pending_candidates": 10000}, + "expected": "include", + "rate": [{"name": "grp", "scope": "global", "payload_types": [5], "window": "5m", "threshold": 20` + g + `, + "state": {"confirm_after": "0s", "end_after_quiet": "5m", "cooldown": "0s"}}]}`)) + if err != nil { + t.Fatal(err) + } + rep, err := anomaly.Replay(bytes.NewReader(in), cfg, anomaly.ReplayOptions{Start: fxT, End: fxAt(2 * time.Hour), MaxLineBytes: 4096}) + if err != nil { + t.Fatal(err) + } + var as []time.Duration + for _, c := range rep.Candidates { + if c.To == anomaly.StateActive { + as = append(as, c.At.Sub(fxT)) + } + } + return as + } + str := func(as []time.Duration) string { return fmt.Sprint(as) } + backlog, quiet, skew, iso := ltMin(45, 19), ltMin(55, 19), ltMin(72, 44), ltMin(95, 19) + if a := activations(late, false); str(a) != str([]time.Duration{backlog, quiet, skew, iso}) { + t.Fatalf("unguarded activations %v", a) + } + if a := activations(old, false); str(a) != str([]time.Duration{backlog, quiet, skew, iso}) { + t.Fatalf("unguarded activations on the #473 export %v", a) + } + if a := activations(late, true); str(a) != str([]time.Duration{quiet, skew, iso}) { + t.Fatalf("late-delivery guarded activations %v, want the three real storms", a) + } + if a := activations(old, true); len(a) != 0 { + t.Fatalf("#473 guarded activations %v, want none", a) + } +} + +// The checked-in E2E fixture (older schema, one observation per +// transmission) exports with the flag: nothing is already seen, nothing is +// flagged, and the output equals the plain export. +func TestExportLateDeliveryE2EFixture(t *testing.T) { + src := filepath.Join("..", "..", "test-fixtures", "e2e-fixture.db") + if _, err := os.Stat(src); err != nil { + t.Skip("e2e fixture not present") + } + before := fileSum(t, src) + opt := exportOptions{DB: src, Start: time.Date(2026, 5, 17, 0, 0, 0, 0, time.UTC), + End: time.Date(2026, 5, 18, 0, 0, 0, 0, time.UTC), Settle: 30 * time.Second, MaxRows: 100000} + plain, _ := runExportTo(t, opt) + opt.LateWindow, opt.LateMinRows, opt.LateShare, opt.LateGap = ltWindow, ltMinRows, ltShare, ltGap + got, sum := runExportTo(t, opt) + if fileSum(t, src) != before { + t.Fatal("e2e fixture changed") + } + if l := sum.LateDelivery; !bytes.Equal(got, plain) || l == nil || l.Rows < 400 || l.RowsAlreadySeen != 0 || + l.EventsFlagged != 0 || sum.EventsWritten < 400 { + t.Fatalf("export differs, or summary %+v (%d events)", sum.LateDelivery, sum.EventsWritten) + } +} diff --git a/cmd/anomaly/late_sweep_538_test.go b/cmd/anomaly/late_sweep_538_test.go new file mode 100644 index 000000000..44db5ca74 --- /dev/null +++ b/cmd/anomaly/late_sweep_538_test.go @@ -0,0 +1,222 @@ +package main + +import ( + "fmt" + "math/rand/v2" + "sort" + "testing" + "time" +) + +// Issue #538: the late-delivery flag on randomised synthetic networks, +// through the real export. Each seed builds 8 observers, two of them with +// clocks 3-10 minutes behind; ADVERT background every 30 s; two real +// GRP_TXT storms of 30-150 transmissions over 20 s - 5 min with a random +// hearing probability; and one observer outage of 5-40 minutes, after +// which that observer delivers its queue (its rows of the outage plus 10-40 +// transmissions it alone heard) stamped with the delivery time. +// +// Each transmission's class is known from the generator and written into +// its payload size, so a line's payload_bytes tells whether its event is +// live (first observation on time; skewed clocks included) or late (its +// first observation is a delivered queue row). A live event whose first +// observation is the draining observer's, within two windows of its +// delivery, has a class of its own: the flag marks whole bursts, so it may +// mark those. + +const ( + swLiveAdvert = 100 + swNearAdvert = 95 + swLateAdvert = 90 + swLiveGrp = 23 + swNearGrp = 28 + swLateGrp = 33 +) + +type swObs struct { + observer int + at int64 // seconds from fxT + late bool // a queue row stamped at delivery +} + +type swTx struct { + pt int + obs []swObs + late bool // the event's first observation is a queue row + near bool // live, first observed by the draining observer near its delivery +} + +type swCase struct { + outage, queue int // minutes, unique queued transmissions + txs []swTx +} + +func sweepCase(seed uint64) swCase { + r := rand.New(rand.NewPCG(seed, 538)) + const nObs = 8 + offset := make([]int64, nObs) + offset[0] = -int64(180 + r.IntN(421)) + offset[1] = -int64(180 + r.IntN(421)) + drain := 2 + r.IntN(nObs-2) + s0 := int64(300 + r.IntN(3301)) + c := swCase{outage: 5 + r.IntN(36), queue: 10 + r.IntN(31)} + rc := s0 + int64(c.outage)*60 + var queued []*swObs + hear := func(pt int, at int64, p float64) { + var obs []swObs + for len(obs) == 0 { + for o := 0; o < nObs; o++ { + if r.Float64() < p { + obs = append(obs, swObs{observer: o, at: at + int64(r.IntN(21)) + offset[o]}) + } + } + } + c.txs = append(c.txs, swTx{pt: pt, obs: obs}) + } + for at := int64(-150 * 60); at < 140*60; at += 30 { + hear(4, at+int64(r.IntN(30)), 0.5) + } + for s := 0; s < 2; s++ { + start, n, span, p := int64(600+r.IntN(6001)), 30+r.IntN(121), 20+r.IntN(281), 0.1+0.8*r.Float64() + for i := 0; i < n; i++ { + hear(5, start+int64(r.IntN(span)), p) + } + } + for i := 0; i < c.queue; i++ { + c.txs = append(c.txs, swTx{pt: 5, obs: []swObs{{observer: drain, at: rc}}}) + } + for i := range c.txs { + for j := range c.txs[i].obs { + if o := &c.txs[i].obs[j]; o.observer == drain && o.at >= s0 && o.at <= rc { + queued = append(queued, o) + } + } + } + r.Shuffle(len(queued), func(i, j int) { queued[i], queued[j] = queued[j], queued[i] }) + for j, o := range queued { + o.at, o.late = rc+int64(j/2), true + } + for i := range c.txs { + x := &c.txs[i] + sort.SliceStable(x.obs, func(a, b int) bool { return x.obs[a].at < x.obs[b].at }) + x.late = x.obs[0].late + x.near = !x.late && x.obs[0].observer == drain && x.obs[0].at >= rc-2*60 && x.obs[0].at < rc+int64(c.queue+len(queued))/2+3*60 + } + return c +} + +func buildSweepFixture(t *testing.T, c swCase) *fixtureDB { + f := newFixtureDB(t, false) + if _, err := f.db.Exec(`PRAGMA synchronous = OFF`); err != nil { + t.Fatal(err) + } + tx, err := f.db.Begin() + if err != nil { + t.Fatal(err) + } + for i, x := range c.txs { + size := map[[2]bool]int{{false, false}: swLiveAdvert, {false, true}: swLateAdvert, + {true, false}: swLiveGrp, {true, true}: swLateGrp}[[2]bool{x.pt == 5, x.late}] + if x.near { + size = map[bool]int{false: swNearAdvert, true: swNearGrp}[x.pt == 5] + } + payload := make([]byte, size) + if x.pt == 5 { + payload = grpPayload(0x0a, size-3) + } + raw := frameHex(1, x.pt, 1, []byte{0x0b}, payload) + res, err := tx.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, payload_type) VALUES (?, ?, '', ?)`, + raw, fxHash(100000+i), x.pt) + if err != nil { + t.Fatal(err) + } + id, _ := res.LastInsertId() + for _, o := range x.obs { + if _, err := tx.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, raw_hex) + VALUES (?, ?, '["0B"]', ?, ?)`, id, 82001+o.observer, fxT.Unix()+o.at, raw); err != nil { + t.Fatal(err) + } + } + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + return f +} + +// swCount is flagged/all events of the live, near and late classes. +type swCount struct{ liveFlag, live, nearFlag, near, lateFlag, late int } + +func countSweep(t *testing.T, out []byte) swCount { + t.Helper() + var c swCount + for _, r := range readRecords(t, out) { + if r.PayloadBytes == nil { + t.Fatal("line without payload_bytes") + } + switch *r.PayloadBytes { + case swLiveAdvert, swLiveGrp: + c.live++ + if r.IngestBacklog { + c.liveFlag++ + } + case swNearAdvert, swNearGrp: + c.near++ + if r.IngestBacklog { + c.nearFlag++ + } + case swLateAdvert, swLateGrp: + c.late++ + if r.IngestBacklog { + c.lateFlag++ + } + default: + t.Fatalf("payload_bytes %d", *r.PayloadBytes) + } + } + return c +} + +// On every seed the late-delivery flag marks no live event, also with two +// skewed clocks, and it marks most delivered-queue events; the #473 flag +// marks live events on every seed. +func TestLateDeliverySyntheticSweep(t *testing.T) { + const seeds = 20 + var late, old swCount + for seed := uint64(1); seed <= seeds; seed++ { + c := sweepCase(seed) + f := buildSweepFixture(t, c) + opt := ltOptions(f.path) + opt.Settle, opt.MaxRows = 30*time.Second, 100000 + got, _ := runExportTo(t, opt) + l := countSweep(t, got) + opt = fxOptions(f.path) + opt.Settle, opt.MaxRows = 30*time.Second, 100000 + opt.BacklogSilence, opt.BacklogSpan = 10*time.Minute, 2*time.Hour + got, _ = runExportTo(t, opt) + o := countSweep(t, got) + t.Logf("seed %2d outage %2dm queue %2d | live %4d: late-delivery %d, #473 %3d | near %d: %d, %d | late %3d: late-delivery %3d, #473 %3d", + seed, c.outage, c.queue, l.live, l.liveFlag, o.liveFlag, l.near, l.nearFlag, o.nearFlag, l.late, l.lateFlag, o.lateFlag) + if l.liveFlag != 0 { + t.Errorf("seed %d: %d live events flagged", seed, l.liveFlag) + } + if o.liveFlag == 0 || l.live != o.live || l.late != o.late || l.late < c.queue { + t.Errorf("seed %d: counts %+v / %+v", seed, l, o) + } + for _, p := range []*swCount{&late, &old} { + q := l + if p == &old { + q = o + } + p.liveFlag, p.live, p.lateFlag, p.late = p.liveFlag+q.liveFlag, p.live+q.live, p.lateFlag+q.lateFlag, p.late+q.late + p.nearFlag, p.near = p.nearFlag+q.nearFlag, p.near+q.near + } + } + share := func(a, b int) string { return fmt.Sprintf("%d/%d (%.1f %%)", a, b, 100*float64(a)/float64(b)) } + t.Logf("all seeds: live flagged: late-delivery %s, #473 %s", share(late.liveFlag, late.live), share(old.liveFlag, old.live)) + t.Logf("all seeds: near flagged: late-delivery %s, #473 %s", share(late.nearFlag, late.near), share(old.nearFlag, old.near)) + t.Logf("all seeds: late flagged: late-delivery %s, #473 %s", share(late.lateFlag, late.late), share(old.lateFlag, old.late)) + if 2*late.lateFlag < late.late { + t.Errorf("late-delivery flag marks only %s of the delivered-queue events", share(late.lateFlag, late.late)) + } +} diff --git a/cmd/anomaly/main.go b/cmd/anomaly/main.go index 7d7f554b8..66abf1cee 100644 --- a/cmd/anomaly/main.go +++ b/cmd/anomaly/main.go @@ -3,7 +3,7 @@ // ingestor and is not built into any image. // // corescope-anomaly export -db PATH [-immutable] -start T -end T -settle D -max-rows N -out FILE -// [-backlog-silence D -backlog-span D] +// [-backlog-silence D -backlog-span D | -late-window D -late-min-rows N -late-share F -late-gap D] // corescope-anomaly replay -config FILE -input FILE|- -start T -end T -max-line-bytes N [-out FILE] // corescope-anomaly keys -input FILE|- -payload-type N -top N [-exclude-channels LIST] -max-keys N -max-line-bytes N [-out FILE] // corescope-anomaly inject -spec FILE -input FILE|- -out FILE -labels FILE -max-injected N -max-line-bytes N @@ -38,7 +38,7 @@ import ( const usage = `usage: corescope-anomaly export -db PATH [-immutable] -start RFC3339 -end RFC3339 -settle DURATION -max-rows N -out FILE - [-backlog-silence DURATION -backlog-span DURATION] + [-backlog-silence DURATION -backlog-span DURATION | -late-window DURATION -late-min-rows N -late-share F -late-gap DURATION] corescope-anomaly replay -config FILE -input FILE|- -start RFC3339 -end RFC3339 -max-line-bytes N [-out FILE] corescope-anomaly keys -input FILE|- -payload-type N -top N [-exclude-channels LIST] -max-keys N -max-line-bytes N [-out FILE] corescope-anomaly inject -spec FILE -input FILE|- -out FILE -labels FILE -max-injected N -max-line-bytes N From bdabcc8764f56d592cfaa593801d1ba021a0600a Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 09:44:00 +0100 Subject: [PATCH 2/5] anomaly: test the late-delivery gap bound with a second delivery (#538) Relates to #538 Co-Authored-By: Claude Opus 5.5 --- cmd/anomaly/late.go | 5 +++-- cmd/anomaly/late_538_test.go | 6 +++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/cmd/anomaly/late.go b/cmd/anomaly/late.go index 93fa2dbc0..c0de484e5 100644 --- a/cmd/anomaly/late.go +++ b/cmd/anomaly/late.go @@ -93,11 +93,12 @@ func (o exportOptions) validateLate() error { // lateQuery selects the observer, time and elsewhere time e (NULL when // there is none) of every observation in [?1, ?2), in time order, at most -// ?3 rows; ?4 is the gap in seconds. +// ?3 rows; ?4 is the gap in seconds. "<>" is never true for a NULL +// observer, so rows without one are never "another observer". const lateQuery = `SELECT o.observer_idx, o.timestamp, (SELECT MAX(p.timestamp) FROM observations p WHERE p.transmission_id = o.transmission_id AND p.timestamp < o.timestamp + ?4 - AND p.observer_idx IS NOT NULL AND p.observer_idx <> o.observer_idx) + AND p.observer_idx <> o.observer_idx) FROM observations o WHERE o.timestamp >= ?1 AND o.timestamp < ?2 ORDER BY o.timestamp, o.id diff --git a/cmd/anomaly/late_538_test.go b/cmd/anomaly/late_538_test.go index 6b883cd61..7cc1670b0 100644 --- a/cmd/anomaly/late_538_test.go +++ b/cmd/anomaly/late_538_test.go @@ -39,6 +39,7 @@ const ( ltQuiet = 81004 // silent, then first to hear a real storm (the #473 loss) ltSkew = 81005 // clock 167 s behind: its rows look early, everyone else's late ltIsolated = 81006 // silent, then a real storm nobody else hears + ltAgain = 81007 // delivers Drain's outage transmissions once more, an hour later ) // ltGroup names the part of the scenario a transmission belongs to. @@ -78,7 +79,10 @@ func lateScenario() []ltTx { obs := []blObs{{ltA, 0}, {ltB, time.Second}} switch { case m >= 5 && m < 45: - obs = append(obs, blObs{ltDrain, blAt(ltMin(45, m-5)).Sub(at)}) + // a later delivery by another observer does not hide that A + // and B heard it long before Drain: only observations before + // t + gap count + obs = append(obs, blObs{ltDrain, blAt(ltMin(45, m-5)).Sub(at)}, blObs{ltAgain, blAt(ltMin(105, m-5)).Sub(at)}) case m != 45: obs = append(obs, blObs{ltDrain, 2 * time.Second}) } From fca73b3abfe4f129b6e3cba94894065aed6b0100 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 09:48:00 +0100 Subject: [PATCH 3/5] anomaly: rename observers with a parity-changing shift in the #538 privacy test Relates to #538 Co-Authored-By: Claude Opus 5.5 --- cmd/anomaly/late_538_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cmd/anomaly/late_538_test.go b/cmd/anomaly/late_538_test.go index 7cc1670b0..4a3c81d3d 100644 --- a/cmd/anomaly/late_538_test.go +++ b/cmd/anomaly/late_538_test.go @@ -304,6 +304,8 @@ func TestExportLateDeliveryIsInvariantUnderObserverRenaming(t *testing.T) { for _, q := range []string{ `UPDATE observations SET observer_idx = 900000 - observer_idx WHERE observer_idx IS NOT NULL`, `UPDATE observations SET observer_idx = CASE observer_idx WHEN 818999 THEN 7 WHEN 818997 THEN 3 ELSE observer_idx END`, + // a shift by an odd number: every index changes parity + `UPDATE observations SET observer_idx = observer_idx + 1000001 WHERE observer_idx IS NOT NULL`, } { if _, err := f.db.Exec(q); err != nil { t.Fatal(err) @@ -314,7 +316,7 @@ func TestExportLateDeliveryIsInvariantUnderObserverRenaming(t *testing.T) { } } var n int - f.db.QueryRow(`SELECT COUNT(*) FROM observations WHERE observer_idx IN (3, 7)`).Scan(&n) + f.db.QueryRow(`SELECT COUNT(*) FROM observations WHERE observer_idx IN (1000004, 1000008)`).Scan(&n) if n == 0 { t.Fatal("the renaming touched no rows") } From c428a80b3590399e94bded0564ede5d7d75d737c Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 09:44:23 +0000 Subject: [PATCH 4/5] anomaly: pin the late-delivery limits and the exclusive t + gap bound (#538) Review round 2 of #591 (F3-F6). The tests pin behaviour the definition already has, so a change to it is a visible decision: - a common-cause outage (two or three observers draining the same outage together) is never flagged, while one draining observer is; - recall of a 40-GRP_TXT FIFO queue depends on the shared rows older than the gap that come with it (0/40 after 3-10 min, 40/40 after 20-40 min at 6 shared rows a minute, 0/40 at 1 a minute); - two clocks behind by different offsets flag 13 of 40 events of a live storm at share 0.5; - a peer exactly at t + gap is not within the gap (kills the reviewer's mutant X1, < to <=). Co-Authored-By: Claude Opus 5.5 --- cmd/anomaly/late_limits_538_test.go | 218 ++++++++++++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 cmd/anomaly/late_limits_538_test.go diff --git a/cmd/anomaly/late_limits_538_test.go b/cmd/anomaly/late_limits_538_test.go new file mode 100644 index 000000000..2a4627cb2 --- /dev/null +++ b/cmd/anomaly/late_limits_538_test.go @@ -0,0 +1,218 @@ +package main + +import ( + "database/sql" + "testing" + "time" +) + +// Issue #538: the stated limits of the late-delivery flag, pinned so that a +// change to them is a visible decision. Each network below is built with the +// sweep's fixture and classes (late: the event's first observation is a +// delivered queue row; live: everything else). + +// limitNet describes a network for the limits: observers 0 and 1 are live +// and hear an ADVERT every period seconds (1 at +1 s); each drainer +// (observers 2, 3, ...) hears it at +2 s, except during the outage, when it +// queues it. queue GRP_TXT heard only by the drainers are spread evenly +// over the outage. On reconnect each drainer delivers its queue in the +// order it heard it (FIFO), rate rows a second, stamped at delivery; +// drainer j starts j seconds after the first. +type limitNet struct { + period, outage, queue, drainers, rate int // outage in minutes +} + +func (n limitNet) build() swCase { + const s0 = 30 * 60 + rc := int64(s0 + n.outage*60) + c := swCase{outage: n.outage, queue: n.queue} + type heard struct { + tx, at int64 // index into c.txs and time heard + } + var queued []heard + for at := int64(-60 * 60); at < 140*60; at += int64(n.period) { + obs := []swObs{{observer: 0, at: at}, {observer: 1, at: at + 1}} + if at >= s0 && at < rc { + queued = append(queued, heard{int64(len(c.txs)), at}) + } else { + for j := 0; j < n.drainers; j++ { + obs = append(obs, swObs{observer: 2 + j, at: at + 2}) + } + } + c.txs = append(c.txs, swTx{pt: 4, obs: obs}) + } + for i := 0; i < n.queue; i++ { + at := int64(s0) + (2*int64(i)+1)*int64(n.outage)*30/int64(n.queue) + queued = append(queued, heard{int64(len(c.txs)), at}) + c.txs = append(c.txs, swTx{pt: 5, late: true}) + } + // FIFO: in the order heard; ties keep the background first + for i := 1; i < len(queued); i++ { + for k := i; k > 0 && queued[k].at < queued[k-1].at; k-- { + queued[k], queued[k-1] = queued[k-1], queued[k] + } + } + for q, h := range queued { + for j := 0; j < n.drainers; j++ { + x := &c.txs[h.tx] + x.obs = append(x.obs, swObs{observer: 2 + j, at: rc + int64(j) + int64(q/n.rate), late: true}) + } + } + return c +} + +// limitFlags exports the network with the PR's flag values and counts the +// flagged events of each class. +func limitFlags(t *testing.T, n limitNet) swCount { + t.Helper() + f := buildSweepFixture(t, n.build()) + opt := ltOptions(f.path) + opt.Settle, opt.MaxRows = 30*time.Second, 100000 + got, _ := runExportTo(t, opt) + return countSweep(t, got) +} + +// A common-cause outage is never flagged: when two or more observers lose +// the same uplink and drain the same outage together, each queued row has a +// peer (the other draining observer) seconds away, so the latest other +// observation before t + gap is that peer, no row is already seen and +// nothing is flagged. One observer draining the same outage is flagged. +func TestLateDeliveryMissesACommonCauseOutage(t *testing.T) { + for _, n := range []limitNet{ + {period: 60, outage: 40, queue: 25, rate: 1}, + {period: 10, outage: 20, queue: 40, rate: 1}, + {period: 10, outage: 40, queue: 40, rate: 1}, + } { + for d := 1; d <= 3; d++ { + n.drainers = d + c := limitFlags(t, n) + t.Logf("background %2ds outage %2dm queue %d drainers %d: late %d/%d, live %d/%d", + n.period, n.outage, n.queue, d, c.lateFlag, c.late, c.liveFlag, c.live) + want := 0 + if d == 1 { + want = n.queue + } + if c.late != n.queue || c.lateFlag != want || c.liveFlag != 0 { + t.Errorf("%+v: late %d/%d, live %d flagged; want %d/%d late, no live", + n, c.lateFlag, c.late, c.liveFlag, want, n.queue) + } + } + } +} + +// Recall at a queue large enough to matter to the GRP_TXT rate rule: the +// share's denominator holds the queue-only rows, so a queue of 40 GRP_TXT +// (delivered FIFO at 5 rows a second) is flagged only when at least as many +// shared rows older than the gap come with it: after a 20- or 40-minute +// outage at 6 shared rows a minute, not after 3-10 minutes, and never at 1 +// shared row a minute. +func TestLateDeliveryRecallOfALargeQueue(t *testing.T) { + for _, c := range []struct { + period int // seconds between shared rows + outage []int + want int + }{ + {10, []int{3, 5, 10}, 0}, + {10, []int{20, 40}, 40}, + {60, []int{3, 5, 10, 20, 40}, 0}, + } { + for _, m := range c.outage { + n := limitNet{period: c.period, outage: m, queue: 40, drainers: 1, rate: 5} + got := limitFlags(t, n) + t.Logf("shared every %2ds outage %2dm: late %d/%d, live %d/%d", c.period, m, got.lateFlag, got.late, got.liveFlag, got.live) + if got.late != 40 || got.lateFlag != c.want || got.liveFlag != 0 { + t.Errorf("%+v: late %d/%d, live %d flagged; want %d/40 late, no live", n, got.lateFlag, got.late, got.liveFlag, c.want) + } + } + } +} + +// Share 0.5 does not protect a real storm from two skewed clocks: a storm of +// 40 GRP_TXT heard by O (clock right), a third of them also by an observer +// whose clock runs 3 minutes behind and another third by one 7 minutes +// behind, and by nobody else, gives two thirds of O's rows delays of 3 or 7 +// minutes. Those span more than the gap, so the 13 events O heard first +// (the ones neither skewed observer heard) are flagged. +func TestLateDeliveryTwoSkewedClocksDefeatTheShare(t *testing.T) { + var c swCase + for at := int64(-60 * 60); at < 140*60; at += 30 { + c.txs = append(c.txs, swTx{pt: 4, obs: []swObs{{observer: 3, at: at}, {observer: 4, at: at + 1}}}) + } + const start = 60 * 60 + for i := int64(0); i < 40; i++ { + obs := []swObs{{observer: 0, at: start + 3*i}} + switch i % 3 { + case 0: + obs = append(obs, swObs{observer: 1, at: start + 3*i - 180}) + case 1: + obs = append(obs, swObs{observer: 2, at: start + 3*i - 420}) + } + // a live storm: any flag on it is wrong, so its class is live + c.txs = append(c.txs, swTx{pt: 5, obs: obs}) + } + for i := range c.txs { + x := &c.txs[i] + for k := 1; k < len(x.obs); k++ { + for j := k; j > 0 && x.obs[j].at < x.obs[j-1].at; j-- { + x.obs[j], x.obs[j-1] = x.obs[j-1], x.obs[j] + } + } + } + f := buildSweepFixture(t, c) + opt := ltOptions(f.path) + opt.Settle, opt.MaxRows = 30*time.Second, 100000 + got, _ := runExportTo(t, opt) + n := countSweep(t, got) + t.Logf("storm and background: live %d/%d flagged", n.liveFlag, n.live) + if n.liveFlag != 13 { + t.Fatalf("%d live events flagged, want the 13 storm events O heard first", n.liveFlag) + } +} + +// The t + gap bound of the elsewhere time is exclusive: a peer exactly at +// t + gap is not "within the gap", so an older peer still makes the row +// already seen; a peer one second earlier is within the gap. +func TestLateQueryPeerAtExactlyTheGapIsNotWithin(t *testing.T) { + const t0, o, p, q = 60 * 60, 81001, 81002, 81003 + gap := int64(ltGap / time.Second) + f := buildFixtureOf(t, []blTx{ + {at: fxAt(t0 * time.Second), pt: 5, obs: []blObs{{o, 0}, {p, -3 * time.Minute}, {q, ltGap}}}, + {at: fxAt((t0 + 600) * time.Second), pt: 5, obs: []blObs{{o, 0}, {p, -3 * time.Minute}, {q, ltGap - time.Second}}}, + }) + db, err := openReadOnly(f.path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + rows, err := db.Query(lateQuery, fxT.Unix(), fxAt(2*time.Hour).Unix(), 100, gap) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + got := map[int64]sql.NullInt64{} + for rows.Next() { + var observer, elsewhere sql.NullInt64 + var ts int64 + if err := rows.Scan(&observer, &ts, &elsewhere); err != nil { + t.Fatal(err) + } + if observer.Int64 == o { + got[ts-fxT.Unix()] = elsewhere + } + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + for ts, want := range map[int64]int64{t0: t0 - 180, t0 + 600: t0 + 600 + gap - 1} { + if e := got[ts]; !e.Valid || e.Int64-fxT.Unix() != want { + t.Fatalf("row at %d s: elsewhere %+v, want %d s", ts, e, want) + } + } + l := ltTracker(1, 100) + for ts, want := range map[int64]bool{t0: true, t0 + 600: false} { + e := got[ts] + if seen, _ := l.see(o, ts+fxT.Unix(), e); seen != want { + t.Fatalf("row at %d s: already seen %v, want %v", ts, seen, want) + } + } +} From a6433216e091790ceebd535295c76957e609a127 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 09:44:23 +0000 Subject: [PATCH 5/5] anomaly: state the late-delivery limits (#538) Review round 2 of #591 (F3-F5, F7, F8): the README and the code comment state that a common-cause outage is never flagged, that recall falls for large queues after short outages, that no share protects against two clocks skewed by different offsets, that outages shorter than about two gaps are never flagged, and that the share bound is compared in floating point (exact for 0.5). No behaviour change. Co-Authored-By: Claude Opus 5.5 --- cmd/anomaly/README.md | 29 +++++++++++++++++++++++++++++ cmd/anomaly/late.go | 9 ++++++++- 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/cmd/anomaly/README.md b/cmd/anomaly/README.md index 21c19ec1a..12fe852e3 100644 --- a/cmd/anomaly/README.md +++ b/cmd/anomaly/README.md @@ -122,6 +122,35 @@ Unlike the #473 flag, it needs no silence: a real storm that an often-quiet observer hears first is not flagged, because the other observers hear the same transmissions within seconds or not at all. +**Limits.** These are properties of the definition, pinned in +`late_limits_538_test.go` at window 1m, min rows 10, share 0.5 and gap 2m: + +- A **common-cause outage is never flagged.** When two or more observers lose + the same uplink or broker and drain the same outage together, each queued + row has a peer, the other draining observer, seconds away. That peer is + the latest other observation before `t + gap`, so no row is already seen. + One observer draining a 40-min outage is flagged (25 of 25, 40 of 40); two + or three draining it together are not (0). +- **Recall falls for large queues after short outages.** The share counts the + queued rows only the draining observer heard, so a queue is flagged only + when at least as many shared rows older than the gap come with it. A queue + of 40 GRP_TXT, delivered in order at 5 rows/s: 40 of 40 flagged after a 20- + or 40-min outage with 6 shared rows a minute; 0 of 40 after 3-10 min; 0 of + 40 after 3-40 min with 1 shared row a minute. +- **A share does not protect against skewed clocks.** Two observers whose + clocks run behind by different offsets give another observer's rows two + different delays. Their spread can exceed the gap: a storm of 40 GRP_TXT + heard by O and, a third each, by two observers 3 and 7 min behind marks + the 13 events O heard first, at share 0.5. +- An outage shorter than about 2 gap is never flagged: already-seen delays are + at least the gap and must spread over at least the gap. +- A queue from an observer that overlaps no other observer, or a queue that + keeps its receive times, is not flagged. +- The share is compared as `seen >= share * rows` in floating point, so a few + bounds that are exact in decimal are missed (share 0.07 with 7 of 100 rows + seen is not a burst). Shares that are multiples of 1/2^k, such as 0.5, are + exact. + **Key-free.** As for the #473 flag: `observer_idx` stays in memory, the flag is the only change to a line, and the output is byte-identical when observers are renamed. diff --git a/cmd/anomaly/late.go b/cmd/anomaly/late.go index c0de484e5..f9db45b1d 100644 --- a/cmd/anomaly/late.go +++ b/cmd/anomaly/late.go @@ -39,6 +39,11 @@ import ( // sets the event's time) is a row of O in a late burst (O, its bin). // Rows with a NULL observer are never flagged. // +// Limits (README, late_limits_538_test.go): observers that drain a common +// outage together are each other's peer and are never flagged; the share +// counts queue-only rows, so a large queue after a short outage is missed; +// two clocks behind by different offsets can pass the spread at any share. +// // The observer is read as observations.observer_idx and used only inside // the export, as a map key and for "another observer"; only the boolean // leaves, so the output does not change when observers are renamed. @@ -171,7 +176,9 @@ func (l *lateTracker) see(observer, ts int64, elsewhere sql.NullInt64) (bool, er } // flagged reports whether an observation of observer at ts lies in a late -// burst of that observer. +// burst of that observer. The share bound is in floating point: exact for +// shares that are multiples of 1/2^k (0.5), while for example 0.07 * 100 +// rounds above 7. func (l *lateTracker) flagged(observer sql.NullInt64, ts int64) bool { if !observer.Valid { return false