Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .eslintrc.json
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
"MAX_HOP_DIST": "readonly",
"MeshAudio": "readonly",
"MeshConfigReady": "readonly",
"NodeAdverts": "readonly",
"PAYLOAD_COLORS": "readonly",
"PAYLOAD_TYPES": "readonly",
"PERF_SLOW_MS": "readonly",
Expand Down
7 changes: 7 additions & 0 deletions .github/workflows/deploy.yml
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,7 @@ jobs:
node test-coverage-gate.js
node test-node-reach-coverage.js
node test-reach-rank.js
node test-node-adverts.js
node test-issue-1107-live-layout.js
node test-issue-1833-legend-toggle-vcr-offset.js
node test-issue-1532-live-fullscreen.js
Expand Down Expand Up @@ -430,6 +431,11 @@ jobs:
# required shape before the server boots.
run: ./corescope-migrate -db test-fixtures/e2e-fixture.db

- name: Seed route-classified adverts for #2073 E2E
# Runs after the migration because it sets transmissions.route_mask
# (#89). See the file header for the rows and expected classes.
run: sqlite3 test-fixtures/e2e-fixture.db < test-fixtures/seed-2073-route-adverts.sql

- name: Start Go server with fixture DB
run: |
fuser -k 13581/tcp 2>/dev/null || true
Expand Down Expand Up @@ -500,6 +506,7 @@ jobs:
BASE_URL=http://localhost:13581 node test-issue-1146-path-link-contrast-e2e.js 2>&1 | tee -a e2e-output.txt
CHROMIUM_REQUIRE=1 node test-issue-1705-subpath-contrast-e2e.js 2>&1 | tee -a e2e-output.txt
BASE_URL=http://localhost:13581 node test-issue-1147-section-order-e2e.js 2>&1 | tee -a e2e-output.txt
BASE_URL=http://localhost:13581 node test-issue-2073-recent-adverts-e2e.js 2>&1 | tee -a e2e-output.txt
BASE_URL=http://localhost:13581 node test-issue-1151-orphan-separators-e2e.js 2>&1 | tee -a e2e-output.txt
BASE_URL=http://localhost:13581 node test-issue-1486-collapse-reopens-detail-e2e.js 2>&1 | tee -a e2e-output.txt
CHROMIUM_REQUIRE=1 BASE_URL=http://localhost:13581 node test-logo-rebrand-e2e.js 2>&1 | tee -a e2e-output.txt
Expand Down
93 changes: 71 additions & 22 deletions cmd/server/advert_stats.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,28 +16,62 @@ type floodAdvertEntry struct {
hash string
}

// countFloodAdverts counts distinct flood adverts (route_type ==
// advertRouteTypeFlood) whose first-seen lies within the past windowHours. Entries
// with unparseable timestamps are skipped, matching relay-liveness behaviour;
// advertWindow is the exact first-seen window shared by the windowed advert
// counters (flood_advert_count_7d and the #2073 advertCounts). Entries with
// unparseable timestamps are skipped, matching relay-liveness behaviour;
// entries without a hash fall back to their timestamp as the dedup key.
type advertWindow struct {
cutoff time.Time
seen map[string]struct{}
}

func newAdvertWindow(now time.Time, windowHours float64) advertWindow {
return advertWindow{
cutoff: now.Add(-time.Duration(windowHours * float64(time.Hour))),
seen: map[string]struct{}{},
}
}

// admit reports whether an advert first seen at ts lies inside the window
// and has not been counted yet (by hash), and records it.
func (w advertWindow) admit(ts, hash string) bool {
t, ok := parseRelayTS(ts)
if !ok || !t.After(w.cutoff) {
return false
}
key := hash
if key == "" {
key = ts
}
if _, dup := w.seen[key]; dup {
return false
}
w.seen[key] = struct{}{}
return true
}

// advertDateFloor is the SQL pre-filter for an advert window: a DATE-ONLY
// string one day before the window start. A date prefix compares lexically
// the same whatever follows it (separator, precision, zone suffix), so the
// floor never drops an in-window row; the exact check is advertWindow.admit
// in Go, which skips what parseRelayTS cannot parse (e.g. a legacy
// 'YYYY-MM-DD HH:MM:SS' value).
func advertDateFloor(now time.Time, windowHours float64) string {
return now.UTC().Add(-time.Duration(windowHours*float64(time.Hour))).AddDate(0, 0, -1).Format("2006-01-02")
}

// countFloodAdverts counts distinct flood adverts (route_type ==
// advertRouteTypeFlood) whose first-seen lies within the past windowHours
// (see advertWindow).
func countFloodAdverts(entries []floodAdvertEntry, now time.Time, windowHours float64) int {
cutoff := now.Add(-time.Duration(windowHours * float64(time.Hour)))
seen := map[string]struct{}{}
w := newAdvertWindow(now, windowHours)
n := 0
for _, e := range entries {
if e.rt != advertRouteTypeFlood {
continue
}
t, ok := parseRelayTS(e.ts)
if !ok || !t.After(cutoff) {
continue
if e.rt == advertRouteTypeFlood && w.admit(e.ts, e.hash) {
n++
}
key := e.hash
if key == "" {
key = e.ts
}
seen[key] = struct{}{}
}
return len(seen)
return n
}

// CountFloodAdvertsForNode returns how many distinct FLOOD adverts pubkey
Expand All @@ -47,9 +81,17 @@ func countFloodAdverts(entries []floodAdvertEntry, now time.Time, windowHours fl
//
// route_type is filtered in SQL so an advert-spamming node cannot truncate
// the flood count (review feedback on the earlier LIMIT approach). The time
// floor is a DATE-ONLY string with one day of slack: a date prefix compares
// lexically the same across every first_seen format parseRelayTS accepts
// ('T' and ' ' separators alike); the exact window check stays in Go.
// floor is advertDateFloor; the exact window check stays in Go.
//
// This is an external contract (the ArcScope advisor reads it) and is kept
// as is, counted fresh on every request and never cached: route_type 1
// only, i.e. the first-inserted route. It therefore differs from
// advertCounts["7d"].flood (node_advert_routes.go), which is route_mask
// based: that one also counts transport flood (route 0) and never counts a
// mixed advert (flood + zero-hop), while this one counts a mixed advert
// whenever its first-inserted route was 1. When a node-detail request scans
// the node for the #2073 breakdown itself, that scan yields this same number
// (GetNodeAdvertRoutes, pinned by a parity test) and this query is skipped.
//
// The row cap is a pure safety valve on per-request allocation: it applies to
// flood adverts inside the floor window only, and 50000 in ~8 days is ~4 per
Expand All @@ -61,7 +103,14 @@ func countFloodAdverts(entries []floodAdvertEntry, now time.Time, windowHours fl
const floodAdvertRowCap = 50000

func (db *DB) CountFloodAdvertsForNode(pubkey string, windowHours float64, rowCap int) (int, error) {
floor := time.Now().UTC().Add(-time.Duration(windowHours*float64(time.Hour))).AddDate(0, 0, -1).Format("2006-01-02")
return db.countFloodAdvertsForNodeAt(pubkey, windowHours, rowCap, time.Now())
}

// countFloodAdvertsForNodeAt is CountFloodAdvertsForNode at a given instant,
// for the parity test of the #2073 scan that can stand in for it
// (GetNodeAdvertRoutes).
func (db *DB) countFloodAdvertsForNodeAt(pubkey string, windowHours float64, rowCap int, now time.Time) (int, error) {
floor := advertDateFloor(now, windowHours)
rows, err := db.conn.Query(
"SELECT COALESCE(first_seen, ''), COALESCE(route_type, -1), COALESCE(hash, '') FROM transmissions WHERE from_pubkey = ? AND payload_type = ? AND route_type = ? AND first_seen >= ? ORDER BY id DESC LIMIT ?",
pubkey, payloadTypeAdvert, advertRouteTypeFlood, floor, rowCap)
Expand All @@ -77,5 +126,5 @@ func (db *DB) CountFloodAdvertsForNode(pubkey string, windowHours float64, rowCa
}
entries = append(entries, e)
}
return countFloodAdverts(entries, time.Now(), windowHours), nil
return countFloodAdverts(entries, now, windowHours), nil
}
56 changes: 44 additions & 12 deletions cmd/server/db.go
Original file line number Diff line number Diff line change
Expand Up @@ -1248,42 +1248,60 @@ func (db *DB) GetNodeByPubkey(pubkey string) (map[string]interface{}, error) {
// same-name false positives and an adversarial spoof path where any node
// could attribute its transmissions to a victim by naming itself with the
// victim's pubkey. Pubkey is unique by design — that's the whole point.
func (db *DB) GetRecentTransmissionsForNode(pubkey string, limit int) ([]map[string]interface{}, error) {
//
// withRouteClass (node detail with include=advertRoutes, #2073) adds
// route_class to the ADVERT rows; without it the rows are unchanged.
func (db *DB) GetRecentTransmissionsForNode(pubkey string, limit int, withRouteClass bool) (NodeAdvertRows, error) {
if limit <= 0 {
limit = 20
}

selectCols, observerJoin := db.transmissionBaseSQL()

// #1345: order by ingest id, not first_seen (=rxTime). Buffered observer
// uploads with old rxTime would otherwise displace fresh activity from
// the "recent transmissions for node" list.
querySQL := fmt.Sprintf("SELECT %s FROM transmissions t %s WHERE t.from_pubkey = ? ORDER BY t.id DESC LIMIT ?",
selectCols, observerJoin)
args := []interface{}{pubkey, limit}
return db.queryNodeAdvertRows(true, withRouteClass, "t.from_pubkey = ? ORDER BY t.id DESC LIMIT ?", pubkey, limit)
}

// queryNodeAdvertRows runs the transmission-centric query with the given
// WHERE tail (including ORDER BY/LIMIT). With observations it attaches every
// observation (otherwise the rows keep the transmission shape and the best
// observation's fields only: the lean #2073 per-class rows). With
// routeClass it adds, for ADVERTs, route_class (#2073: classifyAdvertRoute
// over route_mask with the route_type fallback); a failed mask read only
// leaves route_class out.
func (db *DB) queryNodeAdvertRows(observations, routeClass bool, whereTail string, args ...interface{}) (NodeAdvertRows, error) {
selectCols, observerJoin := db.transmissionBaseSQL()
querySQL := fmt.Sprintf("SELECT %s FROM transmissions t %s WHERE %s", selectCols, observerJoin, whereTail)

rows, err := db.conn.Query(querySQL, args...)
if err != nil {
return nil, err
}
defer rows.Close()

packets := make([]map[string]interface{}, 0)
packets := make(NodeAdvertRows, 0)
var txIDs []int
for rows.Next() {
p := db.scanTransmissionRow(rows)
if p != nil {
// Placeholder for observations — filled below
p["observations"] = []map[string]interface{}{}
if observations {
// Placeholder for observations — filled below
p["observations"] = []map[string]interface{}{}
}
if id, ok := p["id"].(int); ok {
txIDs = append(txIDs, id)
}
packets = append(packets, p)
}
}
if err := rows.Err(); err != nil {
return nil, err
}
if len(txIDs) == 0 {
return packets, nil
}

// Fetch observations for all transmissions
if len(txIDs) > 0 {
if observations {
obsMap := db.getObservationsForTransmissions(txIDs)
for _, p := range packets {
if id, ok := p["id"].(int); ok {
Expand All @@ -1293,7 +1311,21 @@ func (db *DB) GetRecentTransmissionsForNode(pubkey string, limit int) ([]map[str
}
}
}

if !routeClass {
return packets, nil
}
masks := make(map[int]sql.NullInt64, len(txIDs))
if db.hasRouteMask() {
if err := db.readRouteMasks(txIDs, masks); err != nil {
log.Printf("[node-adverts] route_mask read failed, route_class omitted: %v", err)
return packets, nil
}
}
for _, p := range packets {
if id, ok := p["id"].(int); ok {
setAdvertRouteClass(p, masks[id])
}
}
return packets, nil
}

Expand Down
Loading
Loading