anomaly: backlog signal for late-delivered observations

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.
 

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

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)