diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 676ca18..bdad2fd 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -110,7 +110,7 @@ func main() { log.Fatalf("migrations failed: %v", err) } - store := db.New(pool, resolved.ClockDriftThreshold) + store := db.New(pool, resolved.ClockDriftThreshold, resolved.NodeStaleThreshold) // ── Presence write coalescing ──────────────────────────────────────────── // Ingest writes go through the coalescer; reads keep using the store. @@ -246,7 +246,7 @@ func main() { scheduler := background.New([]background.Task{ background.ViewRefreshTask(store, resolved.ViewRefreshInterval), - background.CleanupTask(store, resolved.TelemetryRetention, resolved.PacketRetention, resolved.CleanupInterval), + background.CleanupTask(store, resolved.TelemetryRetention, resolved.PacketRetention, resolved.NodeDeleteAfter, resolved.CleanupInterval), background.ReconfirmTask(store, resolved.ReconfirmInterval), }) go scheduler.Start(ctx) diff --git a/db/nodes.go b/db/nodes.go index 63dfe7d..30bea8a 100644 --- a/db/nodes.go +++ b/db/nodes.go @@ -127,6 +127,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, s ObserverID: nullableUUID(v.ObserverID), KnownNeighborCount: v.KnownNeighborCount, NeighborIDs: v.NeighborIds, + Stale: v.LastSeen.Valid && v.LastSeen.Time.Before(time.Now().Add(-s.staleThreshold)), } if len(v.Iatas) > 0 { if err := json.Unmarshal(v.Iatas, &node.IATAs); err != nil { @@ -135,7 +136,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, s } } if v.RadioFreqMhz != nil && v.RadioSf != nil && v.RadioBwKhz != nil { - s := fmt.Sprintf("%.1f,%g,%d", *v.RadioFreqMhz, *v.RadioBwKhz, *v.RadioSf) + s := fmt.Sprintf("%g,%g,%d", *v.RadioFreqMhz, *v.RadioBwKhz, *v.RadioSf) node.Radio = &s } items = append(items, node) @@ -170,6 +171,7 @@ func (s *Store) GetNode(ctx context.Context, nodeID uuid.UUID) (*api.Node, error ObserverID: nullableUUID(row.ObserverID), DefaultScope: row.DefaultScopeName, KnownNeighborCount: row.KnownNeighborCount, + Stale: row.LastSeen.Valid && row.LastSeen.Time.Before(time.Now().Add(-s.staleThreshold)), }, LocationSource: row.LocationSource, SupportsMultibytePaths: row.SupportsMultibytePaths, @@ -192,7 +194,7 @@ func (s *Store) GetNode(ctx context.Context, nodeID uuid.UUID) (*api.Node, error } } if row.RadioFreqMhz != nil && row.RadioSf != nil && row.RadioBwKhz != nil { - s := fmt.Sprintf("%.1f,%g,%d", *row.RadioFreqMhz, *row.RadioBwKhz, *row.RadioSf) + s := fmt.Sprintf("%g,%g,%d", *row.RadioFreqMhz, *row.RadioBwKhz, *row.RadioSf) node.Radio = &s } if row.LastAdvertAt.Valid { @@ -288,3 +290,9 @@ func (s *Store) GetNodeNeighbors(ctx context.Context, nodeID uuid.UUID) ([]api.N func (s *Store) ReconfirmNeighbors(ctx context.Context) error { return s.q.ReconfirmNeighbors(ctx) } + +// DeleteOldNodes deletes nodes not seen since the given cutoff. See the DeleteOldNodes SQL +// query for the observer_owners exclusion and known_routes caveat. +func (s *Store) DeleteOldNodes(ctx context.Context, cutoff time.Time) error { + return s.q.DeleteOldNodes(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true}) +} diff --git a/db/nodes_test.go b/db/nodes_test.go index cb50fae..db9a0dd 100644 --- a/db/nodes_test.go +++ b/db/nodes_test.go @@ -242,6 +242,42 @@ func TestListNodes_IATAsUnmarshal(t *testing.T) { } } +func TestListNodes_Stale(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + + staleID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + freshID := uuid.MustParse("00000000-0000-0000-0000-000000000002") + unmeasuredID := uuid.MustParse("00000000-0000-0000-0000-000000000003") + + mock.EXPECT(). + ListNodes(gomock.Any(), gomock.Any()). + Return([]sqlc.ListNodesRow{ + {ID: staleID, PublicKey: []byte{0x01}, LastSeen: pgtype.Timestamptz{Time: time.Now().Add(-48 * time.Hour), Valid: true}}, + {ID: freshID, PublicKey: []byte{0x02}, LastSeen: pgtype.Timestamptz{Time: time.Now(), Valid: true}}, + {ID: unmeasuredID, PublicKey: []byte{0x03}, LastSeen: pgtype.Timestamptz{Valid: false}}, + }, nil) + + store := &Store{q: mock, staleThreshold: 24 * time.Hour} + page, err := store.ListNodes(context.Background(), 0, nil, nil, nil, nil, "", "", "", 0, 10, false) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + byID := make(map[uuid.UUID]bool) + for _, n := range page.Items { + byID[n.ID] = n.Stale + } + if !byID[staleID] { + t.Error("expected node last seen 48h ago to be stale with a 24h threshold") + } + if byID[freshID] { + t.Error("expected node last seen just now to not be stale") + } + if byID[unmeasuredID] { + t.Error("expected a node with no last_seen at all to not be stale") + } +} + func TestListNodes_RadioStringFormatting(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) @@ -271,8 +307,8 @@ func TestListNodes_RadioStringFormatting(t *testing.T) { if page.Items[0].Radio == nil { t.Fatal("expected Radio to be set") } - if *page.Items[0].Radio != "915.0,125,7" { - t.Errorf("expected Radio 915.0,125,7, got %s", *page.Items[0].Radio) + if *page.Items[0].Radio != "915,125,7" { + t.Errorf("expected Radio 915,125,7, got %s", *page.Items[0].Radio) } } @@ -339,6 +375,36 @@ func TestGetNode_LastAdvertAtNil(t *testing.T) { } } +func TestGetNode_Stale(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + + nodeID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + + mock.EXPECT(). + GetNodeByID(gomock.Any(), nodeID). + Return(sqlc.GetNodeByIDRow{ + ID: nodeID, + PublicKey: []byte{0x01}, + NodeType: 1, // companion -- Stale applies to every node type, unlike clock drift + FirstSeen: pgtype.Timestamptz{Time: time.Now().Add(-72 * time.Hour), Valid: true}, + LastSeen: pgtype.Timestamptz{Time: time.Now().Add(-48 * time.Hour), Valid: true}, + }, nil) + + mock.EXPECT(). + GetNodeNeighbors(gomock.Any(), nodeID). + Return([]sqlc.GetNodeNeighborsRow{}, nil) + + store := &Store{q: mock, staleThreshold: 24 * time.Hour} + node, err := store.GetNode(context.Background(), nodeID) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !node.Stale { + t.Error("expected node last seen 48h ago to be stale with a 24h threshold") + } +} + func TestGetNode_ClockDrift_OutOfSync(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) @@ -624,3 +690,19 @@ func TestListNodes_ExcludeNeighbors_LeavesIDsNil(t *testing.T) { t.Errorf("expected NeighborIDs to stay nil when includeNeighbors is false, got %v", page.Items[0].NeighborIDs) } } + +func TestDeleteOldNodes(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + + cutoff := time.Now().Add(-30 * 24 * time.Hour) + + mock.EXPECT(). + DeleteOldNodes(gomock.Any(), gomock.Eq(pgtype.Timestamptz{Time: cutoff, Valid: true})). + Return(nil) + + store := &Store{q: mock} + if err := store.DeleteOldNodes(context.Background(), cutoff); err != nil { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/db/observers.go b/db/observers.go index 744fa68..5ff3706 100644 --- a/db/observers.go +++ b/db/observers.go @@ -61,7 +61,7 @@ func (s *Store) ListObservers(ctx context.Context, iatas []string, observerType, Scopes: v.Scopes, } if v.RadioFreqMhz != nil && v.RadioSf != nil && v.RadioBwKhz != nil { - s := fmt.Sprintf("%.1f,%g,%d", *v.RadioFreqMhz, *v.RadioBwKhz, *v.RadioSf) + s := fmt.Sprintf("%g,%g,%d", *v.RadioFreqMhz, *v.RadioBwKhz, *v.RadioSf) observer.Radio = &s } if v.DisplayName != nil { diff --git a/db/observers_test.go b/db/observers_test.go index 07f738f..f6f7572 100644 --- a/db/observers_test.go +++ b/db/observers_test.go @@ -128,8 +128,8 @@ func TestListObservers_RadioStringFormatting(t *testing.T) { if page.Items[0].Radio == nil { t.Fatal("expected Radio to be set") } - if *page.Items[0].Radio != "915.0,125,7" { - t.Errorf("expected Radio 915.0,125,7, got %s", *page.Items[0].Radio) + if *page.Items[0].Radio != "915,125,7" { + t.Errorf("expected Radio 915,125,7, got %s", *page.Items[0].Radio) } } diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 26b99b7..2cef212 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -509,6 +509,18 @@ LIMIT $6; -- packet_observations cascade-delete via FK. DELETE FROM packets WHERE last_heard_at < $1; +-- name: DeleteOldNodes :exec +-- Deletes nodes not seen since the given cutoff. node_iatas and node_neighbors cascade- +-- delete via FK. Excludes nodes referenced by observer_owners.owner_node_id -- that FK has +-- no ON DELETE action, so deleting one directly would fail the whole statement anyway, and +-- an operator manually recorded ownership for that node, so leave it alone even if stale. +-- known_routes.node_ids is a plain UUID[] with no FK; a deleted node's id can be left +-- dangling in old routes there, but ReconfirmTask already prunes stale/ambiguous routes +-- periodically and will clean those up on its own schedule. +DELETE FROM nodes +WHERE last_seen < $1 + AND id NOT IN (SELECT owner_node_id FROM observer_owners WHERE owner_node_id IS NOT NULL); + -- name: DeleteOldChannelIATAs :exec -- Keeps the channel IATA filter in step with packet retention. DELETE FROM channel_iatas WHERE last_heard < $1; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 36074a3..1cb3b7a 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -57,6 +57,20 @@ func (mr *MockQuerierMockRecorder) DeleteOldChannelIATAs(ctx, lastHeard any) *go return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteOldChannelIATAs", reflect.TypeOf((*MockQuerier)(nil).DeleteOldChannelIATAs), ctx, lastHeard) } +// DeleteOldNodes mocks base method. +func (m *MockQuerier) DeleteOldNodes(ctx context.Context, lastSeen pgtype.Timestamptz) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeleteOldNodes", ctx, lastSeen) + ret0, _ := ret[0].(error) + return ret0 +} + +// DeleteOldNodes indicates an expected call of DeleteOldNodes. +func (mr *MockQuerierMockRecorder) DeleteOldNodes(ctx, lastSeen any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteOldNodes", reflect.TypeOf((*MockQuerier)(nil).DeleteOldNodes), ctx, lastSeen) +} + // DeleteOldPackets mocks base method. func (m *MockQuerier) DeleteOldPackets(ctx context.Context, lastHeardAt pgtype.Timestamptz) error { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 618a1e9..a56029f 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -14,6 +14,14 @@ import ( type Querier interface { // Keeps the channel IATA filter in step with packet retention. DeleteOldChannelIATAs(ctx context.Context, lastHeard pgtype.Timestamptz) error + // Deletes nodes not seen since the given cutoff. node_iatas and node_neighbors cascade- + // delete via FK. Excludes nodes referenced by observer_owners.owner_node_id -- that FK has + // no ON DELETE action, so deleting one directly would fail the whole statement anyway, and + // an operator manually recorded ownership for that node, so leave it alone even if stale. + // known_routes.node_ids is a plain UUID[] with no FK; a deleted node's id can be left + // dangling in old routes there, but ReconfirmTask already prunes stale/ambiguous routes + // periodically and will clean those up on its own schedule. + DeleteOldNodes(ctx context.Context, lastSeen pgtype.Timestamptz) error // Deletes packets and their observations older than the given cutoff. // packet_observations cascade-delete via FK. DeleteOldPackets(ctx context.Context, lastHeardAt pgtype.Timestamptz) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 3b73edb..75c87ee 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -22,6 +22,24 @@ func (q *Queries) DeleteOldChannelIATAs(ctx context.Context, lastHeard pgtype.Ti return err } +const deleteOldNodes = `-- name: DeleteOldNodes :exec +DELETE FROM nodes +WHERE last_seen < $1 + AND id NOT IN (SELECT owner_node_id FROM observer_owners WHERE owner_node_id IS NOT NULL) +` + +// Deletes nodes not seen since the given cutoff. node_iatas and node_neighbors cascade- +// delete via FK. Excludes nodes referenced by observer_owners.owner_node_id -- that FK has +// no ON DELETE action, so deleting one directly would fail the whole statement anyway, and +// an operator manually recorded ownership for that node, so leave it alone even if stale. +// known_routes.node_ids is a plain UUID[] with no FK; a deleted node's id can be left +// dangling in old routes there, but ReconfirmTask already prunes stale/ambiguous routes +// periodically and will clean those up on its own schedule. +func (q *Queries) DeleteOldNodes(ctx context.Context, lastSeen pgtype.Timestamptz) error { + _, err := q.db.Exec(ctx, deleteOldNodes, lastSeen) + return err +} + const deleteOldPackets = `-- name: DeleteOldPackets :exec DELETE FROM packets WHERE last_heard_at < $1 ` diff --git a/db/store.go b/db/store.go index e6ee46a..d2699a7 100644 --- a/db/store.go +++ b/db/store.go @@ -23,13 +23,16 @@ import ( type Store struct { q sqlc.Querier clockDriftThreshold time.Duration // see api.Node.ClockOutOfSync + staleThreshold time.Duration // see api.NodeSummary.Stale } // New creates a Store backed by the given pgxpool connection pool. clockDriftThreshold is // the |device clock - server clock| magnitude above which a repeater/room server's // clockOutOfSync is reported true; see internal/config.ResolvedConfig.ClockDriftThreshold. -func New(pool *pgxpool.Pool, clockDriftThreshold time.Duration) *Store { - return &Store{q: sqlc.New(pool), clockDriftThreshold: clockDriftThreshold} +// staleThreshold is how long since last_seen before a node's Stale is reported true; see +// internal/config.ResolvedConfig.NodeStaleThreshold. +func New(pool *pgxpool.Pool, clockDriftThreshold, staleThreshold time.Duration) *Store { + return &Store{q: sqlc.New(pool), clockDriftThreshold: clockDriftThreshold, staleThreshold: staleThreshold} } func (s *Store) ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { diff --git a/docs/docs.go b/docs/docs.go index 4ce0893..c97b14f 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -4,11 +4,11 @@ package docs import "github.com/swaggo/swag" const docTemplate = `{ - "schemes": {{ marshal .Schemes }}, + "schemes": [[ marshal .Schemes ]], "swagger": "2.0", "info": { - "description": "{{escape .Description}}", - "title": "{{.Title}}", + "description": "[[escape .Description]]", + "title": "[[.Title]]", "termsOfService": "https://github.com/MeshCore-Beacon/beacon-server", "contact": { "name": "MeshCore Beacon", @@ -17,10 +17,10 @@ const docTemplate = `{ "license": { "name": "AGPL-3-or-later" }, - "version": "{{.Version}}" + "version": "[[.Version]]" }, - "host": "{{.Host}}", - "basePath": "{{.BasePath}}", + "host": "[[.Host]]", + "basePath": "[[.BasePath]]", "paths": { "/brokers": { "get": { @@ -291,6 +291,43 @@ const docTemplate = `{ } } }, + "/iatas/{iata}/border": { + "get": { + "produces": [ + "application/json" + ], + "tags": [ + "IATAs" + ], + "summary": "Get an IATA's GeoJSON border, if configured", + "parameters": [ + { + "type": "string", + "description": "3-letter IATA code", + "name": "iata", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "GeoJSON Feature (Polygon or MultiPolygon geometry, with bbox)", + "schema": { + "type": "object" + } + }, + "204": { + "description": "IATA exists but has no border configured" + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, "/messages": { "get": { "produces": [ @@ -2539,6 +2576,10 @@ const docTemplate = `{ "description": "shorthand: \"freqMhz,bwKhz,sf\" e.g. \"910.525,62.5,7\"", "type": "string" }, + "stale": { + "description": "Stale is true when the node hasn't been seen (last_seen) within the configured\nstaleness window (default 24h; internal/config.ResolvedConfig.NodeStaleThreshold).\nApplies to every node type, unlike ClockDriftSeconds/ClockOutOfSync on Node, which\nare repeater/room-server only.", + "type": "boolean" + }, "supportsMultibytePaths": { "description": "firmware \u003e= 1.14.0; detected via path hash size", "type": "boolean" @@ -2664,6 +2705,10 @@ const docTemplate = `{ "radio": { "description": "shorthand: \"freqMhz,bwKhz,sf\" e.g. \"910.525,62.5,7\"", "type": "string" + }, + "stale": { + "description": "Stale is true when the node hasn't been seen (last_seen) within the configured\nstaleness window (default 24h; internal/config.ResolvedConfig.NodeStaleThreshold).\nApplies to every node type, unlike ClockDriftSeconds/ClockOutOfSync on Node, which\nare repeater/room-server only.", + "type": "boolean" } } }, @@ -3774,8 +3819,8 @@ var SwaggerInfo = &swag.Spec{ Description: "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.", InfoInstanceName: "swagger", SwaggerTemplate: docTemplate, - LeftDelim: "{{", - RightDelim: "}}", + LeftDelim: "[[", + RightDelim: "]]", } func init() { diff --git a/docs/swagger.json b/docs/swagger.json index 92312e1..e1b3a3c 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -289,6 +289,43 @@ } } }, + "/iatas/{iata}/border": { + "get": { + "produces": [ + "application/json" + ], + "tags": [ + "IATAs" + ], + "summary": "Get an IATA's GeoJSON border, if configured", + "parameters": [ + { + "type": "string", + "description": "3-letter IATA code", + "name": "iata", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "GeoJSON Feature (Polygon or MultiPolygon geometry, with bbox)", + "schema": { + "type": "object" + } + }, + "204": { + "description": "IATA exists but has no border configured" + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, "/messages": { "get": { "produces": [ @@ -2537,6 +2574,10 @@ "description": "shorthand: \"freqMhz,bwKhz,sf\" e.g. \"910.525,62.5,7\"", "type": "string" }, + "stale": { + "description": "Stale is true when the node hasn't been seen (last_seen) within the configured\nstaleness window (default 24h; internal/config.ResolvedConfig.NodeStaleThreshold).\nApplies to every node type, unlike ClockDriftSeconds/ClockOutOfSync on Node, which\nare repeater/room-server only.", + "type": "boolean" + }, "supportsMultibytePaths": { "description": "firmware \u003e= 1.14.0; detected via path hash size", "type": "boolean" @@ -2662,6 +2703,10 @@ "radio": { "description": "shorthand: \"freqMhz,bwKhz,sf\" e.g. \"910.525,62.5,7\"", "type": "string" + }, + "stale": { + "description": "Stale is true when the node hasn't been seen (last_seen) within the configured\nstaleness window (default 24h; internal/config.ResolvedConfig.NodeStaleThreshold).\nApplies to every node type, unlike ClockDriftSeconds/ClockOutOfSync on Node, which\nare repeater/room-server only.", + "type": "boolean" } } }, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 98154ca..0580247 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -248,6 +248,13 @@ definitions: radio: description: 'shorthand: "freqMhz,bwKhz,sf" e.g. "910.525,62.5,7"' type: string + stale: + description: |- + Stale is true when the node hasn't been seen (last_seen) within the configured + staleness window (default 24h; internal/config.ResolvedConfig.NodeStaleThreshold). + Applies to every node type, unlike ClockDriftSeconds/ClockOutOfSync on Node, which + are repeater/room-server only. + type: boolean supportsMultibytePaths: description: firmware >= 1.14.0; detected via path hash size type: boolean @@ -336,6 +343,13 @@ definitions: radio: description: 'shorthand: "freqMhz,bwKhz,sf" e.g. "910.525,62.5,7"' type: string + stale: + description: |- + Stale is true when the node hasn't been seen (last_seen) within the configured + staleness window (default 24h; internal/config.ResolvedConfig.NodeStaleThreshold). + Applies to every node type, unlike ClockDriftSeconds/ClockOutOfSync on Node, which + are repeater/room-server only. + type: boolean type: object github_com_MeshCore-Beacon_beacon-server_internal_api.NodeTypeCount: properties: @@ -1253,6 +1267,30 @@ paths: summary: Get a single IATA code tags: - IATAs + /iatas/{iata}/border: + get: + parameters: + - description: 3-letter IATA code + in: path + name: iata + required: true + type: string + produces: + - application/json + responses: + "200": + description: GeoJSON Feature (Polygon or MultiPolygon geometry, with bbox) + schema: + type: object + "204": + description: IATA exists but has no border configured + "404": + description: Not Found + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + summary: Get an IATA's GeoJSON border, if configured + tags: + - IATAs /messages: get: parameters: diff --git a/internal/api/nodes.go b/internal/api/nodes.go index 4870cda..a79280b 100644 --- a/internal/api/nodes.go +++ b/internal/api/nodes.go @@ -48,6 +48,11 @@ type NodeSummary struct { DefaultScope *string `json:"defaultScope,omitempty"` // most recently matched transport scope name e.g. "#bc" KnownNeighborCount int64 `json:"knownNeighborCount"` NeighborIDs []uuid.UUID `json:"neighborIds,omitempty"` // only populated when the list request opts in; see ?neighbors=true + // Stale is true when the node hasn't been seen (last_seen) within the configured + // staleness window (default 24h; internal/config.ResolvedConfig.NodeStaleThreshold). + // Applies to every node type, unlike ClockDriftSeconds/ClockOutOfSync on Node, which + // are repeater/room-server only. + Stale bool `json:"stale"` } // Node is the full node representation including firmware capability flags, diff --git a/internal/background/tasks.go b/internal/background/tasks.go index 94c8109..6a5cf68 100644 --- a/internal/background/tasks.go +++ b/internal/background/tasks.go @@ -44,8 +44,8 @@ func ViewRefreshTask(store *db.Store, interval time.Duration) Task { } } -// CleanupTask returns a Task that prunes old telemetry and packet rows. -func CleanupTask(store *db.Store, telemetryRetention, packetRetention, interval time.Duration) Task { +// CleanupTask returns a Task that prunes old telemetry, packet, and node rows. +func CleanupTask(store *db.Store, telemetryRetention, packetRetention, nodeDeleteAfter, interval time.Duration) Task { return Task{ Name: "cleanup", Interval: interval, @@ -65,6 +65,9 @@ func CleanupTask(store *db.Store, telemetryRetention, packetRetention, interval if err := store.DeleteOldTraceIATAs(ctx, cutoff); err != nil { return err } + if err := store.DeleteOldNodes(ctx, time.Now().Add(-nodeDeleteAfter)); err != nil { + return err + } return nil }, } diff --git a/internal/config/config.go b/internal/config/config.go index 4220068..82c4bb8 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -48,6 +48,11 @@ type ResolvedConfig struct { // node's ADVERT timestamp, above which the node API reports clockOutOfSync=true for // that node. Only meaningful for repeaters/room servers (nodeType 2/3). ClockDriftThreshold time.Duration + + // NodeStaleThreshold and NodeDeleteAfter mirror ClockDriftThreshold's "0 means unset, + // resolve to a default" pattern -- see NodesConfig. + NodeStaleThreshold time.Duration + NodeDeleteAfter time.Duration } // PresenceConfig controls coalescing of presence bookkeeping writes @@ -178,6 +183,13 @@ type NodesConfig struct { // repeater/room server's ADVERT timestamp, above which the node API reports // clockOutOfSync=true for that node. Defaults to 5m if not set. ClockDriftThreshold duration `yaml:"clock_drift_threshold"` + // StaleThreshold is how long since a node's last_seen before the node API reports + // stale=true for it. Defaults to 24h if not set. + StaleThreshold duration `yaml:"stale_threshold"` + // DeleteAfter is how long since a node's last_seen before the cleanup job deletes the + // node entirely. Defaults to the same 30-day default as packets.retention if not set -- + // independently configurable from it, just the same starting point. + DeleteAfter duration `yaml:"delete_after"` } // duration is a wrapper around time.Duration that supports YAML unmarshalling @@ -299,6 +311,8 @@ func Resolve(cfg *Config) ResolvedConfig { PresencePacketTTL: cfg.Presence.PacketTTL.Duration, ClockDriftThreshold: cfg.Nodes.ClockDriftThreshold.Duration, + NodeStaleThreshold: cfg.Nodes.StaleThreshold.Duration, + NodeDeleteAfter: cfg.Nodes.DeleteAfter.Duration, } if r.TelemetryResolution == 0 { r.TelemetryResolution = time.Hour @@ -330,14 +344,23 @@ func Resolve(cfg *Config) ResolvedConfig { if r.ClockDriftThreshold == 0 { r.ClockDriftThreshold = 5 * time.Minute } + if r.NodeStaleThreshold == 0 { + r.NodeStaleThreshold = 24 * time.Hour + } + if r.NodeDeleteAfter == 0 { + // Same default as packets.retention (30 days) -- independently configurable, just + // the same starting point, not tied to whatever PacketRetention resolves to. + r.NodeDeleteAfter = 30 * 24 * time.Hour + } return r } func (r ResolvedConfig) String() string { return fmt.Sprintf( - "telemetryResolution=%s telemetryRetention=%s packetRetention=%s maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s presenceFlush=%s presencePacketTTL=%s clockDriftThreshold=%s", + "telemetryResolution=%s telemetryRetention=%s packetRetention=%s maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s presenceFlush=%s presencePacketTTL=%s clockDriftThreshold=%s nodeStaleThreshold=%s nodeDeleteAfter=%s", r.TelemetryResolution, r.TelemetryRetention, r.PacketRetention, r.MaxConnsPerIP, r.ViewRefreshInterval, r.ReconfirmInterval, r.CleanupInterval, r.PresenceFlushInterval, r.PresencePacketTTL, r.ClockDriftThreshold, + r.NodeStaleThreshold, r.NodeDeleteAfter, ) } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 349ba3f..a68e18b 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -98,6 +98,12 @@ func TestResolve_Defaults(t *testing.T) { if r.ClockDriftThreshold != 5*time.Minute { t.Errorf("expected ClockDriftThreshold 5m, got %v", r.ClockDriftThreshold) } + if r.NodeStaleThreshold != 24*time.Hour { + t.Errorf("expected NodeStaleThreshold 24h, got %v", r.NodeStaleThreshold) + } + if r.NodeDeleteAfter != 30*24*time.Hour { + t.Errorf("expected NodeDeleteAfter 720h (same default as PacketRetention), got %v", r.NodeDeleteAfter) + } } func TestResolve_ExplicitValues(t *testing.T) {