diff --git a/cmd/anomaly/README.md b/cmd/anomaly/README.md index c41874b1d..12fe852e3 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,78 @@ 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. + +**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. + +**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 +232,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..f9db45b1d --- /dev/null +++ b/cmd/anomaly/late.go @@ -0,0 +1,235 @@ +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. +// +// 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. +// +// 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. "<>" 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 <> 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. 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 + } + 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..4a3c81d3d --- /dev/null +++ b/cmd/anomaly/late_538_test.go @@ -0,0 +1,651 @@ +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 + ltAgain = 81007 // delivers Drain's outage transmissions once more, an hour later +) + +// 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: + // 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}) + } + 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`, + // 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) + } + 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 (1000004, 1000008)`).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_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) + } + } +} 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