diff --git a/docs/plans/next-gen/README.md b/docs/plans/next-gen/README.md index 2af8a2f..98a5076 100644 --- a/docs/plans/next-gen/README.md +++ b/docs/plans/next-gen/README.md @@ -1,7 +1,9 @@ # Next-Generation Architecture — MeshCore Hub Rewrite -> **Status:** Design complete (iterations 1–7). All 22 architectural decisions locked. All design -> questions resolved. Ready for implementation. +> **Status:** Design complete (iterations 1–8). All 22 architectural decisions locked. All design +> questions resolved. Iteration 8 applied 13 review corrections — see +> [review-findings.md](review-findings.md) — four of them schema-level (multi-tenant uniqueness, CAGG +> vs hypertable, RLS enforcement, phase sequencing) resolved before the Phase 0 DDL freeze. > **Supersedes:** The monolithic `REWRITE.md` (split into these files). This directory contains the complete design for a from-scratch rewrite of MeshCore Hub. @@ -20,6 +22,7 @@ Then read the [decisions](decisions/) for the locked architectural choices, and | Document | Purpose | |---|---| +| [review-findings.md](review-findings.md) | Iteration-8 design-review findings (13 issues) and their resolutions, cross-referenced (F1–F13) into the docs below | | [overview.md](overview.md) | Current system inventory, pain points, target architecture principles, topology | | [code-warts.md](code-warts.md) | Catalog of 52 antipatterns and gotchas from the current codebase (lessons for the new repo) | | [phasing.md](phasing.md) | The 7-phase plan, risks, and design retrospective (what shifted across iterations) | @@ -76,7 +79,7 @@ All 22 decisions are locked. See [decisions/](decisions/) for individual records |---|---|---| | Ingest | Sync MQTT callback, 1 thread | `MqttIngester` → NATS → `IngestWorker` pool | | History tables | `raw_packets` + `packet_path_hops`, 2-day cap | TimescaleDB hypertables, 30-day retention, compressed | -| Route health | 7 tables + 2 background cadences | 3 worker-maintained tables + dashboard CAGGs | +| Route health | 7 tables + 2 background cadences | 3 worker-maintained tables + dashboard CAGGs (packets) & rollup tables (messages/adverts/nodes) | | IDs/enums | `String(36)` UUIDs, string enums | Native `uuid`, Postgres enums, `JSONB` | | Auth | Header injection (implicit trust) | Short-lived JWT + local passwords + optional OIDC | | Cache | Dual/tri key format, hand-coded invalidation | Single key format + declarative dependency graph | diff --git a/docs/plans/next-gen/components/api.md b/docs/plans/next-gen/components/api.md index 85f6814..9f2d6b2 100644 --- a/docs/plans/next-gen/components/api.md +++ b/docs/plans/next-gen/components/api.md @@ -10,6 +10,7 @@ - All handlers are `async` using **Drizzle ORM** over `node-postgres`, which provides native async I/O end-to-end. - DB I/O off the event loop; connection pool sized to async concurrency. - The `@cached` decorator's sync/async branch collapses to one async path. +- **Every request runs inside a transaction that first issues `SET LOCAL app.instance_id` (RLS).** Reads included — `SET LOCAL` is transaction-scoped, so a read outside a transaction sees a NULL GUC and RLS returns 0 rows (data-model.md §1.5, §3.2). A Fastify `preHandler` opens the tx, sets the GUC from the resolved `Principal.instance_id`, and the handler runs inside it. The app connects as the non-owner `meshcore_app` role so `FORCE ROW LEVEL SECURITY` actually applies. ## Error response format @@ -87,12 +88,17 @@ Replaces the dual/tri cache-key format and the hand-coded invalidation helpers. **One key format:** ``` -{namespace}:{scope}:{query_hash} - namespace = endpoint family ("nodes", "messages", "routes", ...) - scope = "shared" | role_tier ("admin", "operator", "member", "community", "anonymous") +{instance_id}:{namespace}:{scope}:{query_hash} + instance_id = the caller's tenant (from the Principal) — REQUIRED so tenants never share a cache entry + namespace = endpoint family ("nodes", "messages", "routes", ...) + scope = "shared" | role_tier ("admin", "operator", "member", "community", "anonymous") query_hash = sha256(sorted_query_params)[:16] ``` +The `instance_id` prefix is not optional: without it two tenants issuing the same query collide on one +cache entry (a cross-tenant data leak), and invalidation cannot be scoped to one tenant. In single-tenant +mode it is a constant prefix; in multi-tenant mode it is what makes the shared Redis safe. + **Declarative invalidation graph** — the single source of truth, replacing AGENTS.md's "hard rule" hand-mapping: ```typescript @@ -124,9 +130,10 @@ def cached(namespace: str, *, ttl_setting: str = "cache_ttl"): @wraps(handler) async def wrapper(request: Request, *args, **kwargs): ns = NAMESPACES[namespace] + iid = request.state.principal.instance_id role = request.state.principal.role_tier if ns.role_scoped else "shared" qhash = sha256(sorted_query_string(request).encode())[:16].hex() - key = f"{namespace}:{role}:{qhash}" + key = f"{iid}:{namespace}:{role}:{qhash}" # Conditional GET → 304 inm = request.headers.get("if-none-match") @@ -160,7 +167,7 @@ async def invalidate_for(session_changes: Iterable[str], cache: CacheBackend, in for entity in session_changes: namespaces |= ENTITY_INVALIDATION.get(entity, set()) for ns in namespaces: - await cache.delete(f"{ns}:*") # SCAN + DEL by prefix + await cache.delete(f"{instance_id}:{ns}:*") # SCAN + DEL by prefix, scoped to THIS tenant only ``` A mutation handler declares what it changed: @@ -180,10 +187,17 @@ A single `applyVisibility(query, principal)` Drizzle query builder construct tha ## Kill the N+1 and the count-subquery -- **Eager-load** observer + tag data with `selectinload` at the ORM level (already done for nodes; extend to events). +- **Eager-load** observer + tag data in one round-trip. In Drizzle this is the relational-queries API (`db.query.messages.findMany({ with: { observers: true, node: { with: { tags: true } } } })`) or an explicit `LEFT JOIN` + row aggregation — there is no `selectinload`; the pattern (avoid per-row hydration), not the SQLAlchemy mechanism, is what carries over. - **Precompute `total`** for hot list endpoints as a denormalized counter or a separate `count` query that doesn't wrap the full filtered query (keyset pagination where possible; for `packet_groups` use a `count(distinct wire_hash)` materialized helper). - **Cache `get_visible_channel_indices`** per request (it's role-scoped and stable) — compute once in the `Principal`. +## Dashboard reads (CAGGs + rollups) + +- The two **CAGGs** (`cagg_daily_packet_counts`, `cagg_packet_breakdown_by_type`, over `raw_receptions`) are read directly, but **RLS does not propagate to a continuous aggregate** — the dashboard query must add an explicit `WHERE instance_id = ` predicate (data-model.md §3.6). +- Daily **message/advert counts** and **node-count history** are read from the worker-maintained rollup tables (`dashboard_daily_message_counts`, `dashboard_daily_advert_counts`, `dashboard_node_count_history`), which *are* RLS-scoped like any tenant table (data-model.md §3.6a). These replace the two message/advert CAGGs that TimescaleDB cannot build over the OLTP tables. + +> **Implementation note (D22).** The pseudocode in this doc is Python-shaped (`@cached` decorator, `Depends`-injected `Principal`/`RequireAdmin`). In the locked TS stack these map to Fastify **`preHandler` hooks** (auth + the per-request transaction that sets `app.instance_id`) and a **cache plugin/hook** rather than a handler decorator — the contract (one key format, one invalidation graph, one resolved `Principal`) is identical; the wiring is Fastify-idiomatic, not decorator-based. + ## OpenAPI as the contract - Enforce a clean OpenAPI schema (already generated). Add Zod response schemas everywhere (some detail endpoints are loose). @@ -289,13 +303,14 @@ Rule of thumb: if changing it requires reconnecting to an external system or re- ```sql CREATE TABLE settings ( - key text PRIMARY KEY, + key text NOT NULL, value jsonb NOT NULL, -- typed per category (validated server-side) category text NOT NULL, -- 'branding' | 'features' | 'tuning' | 'webhooks' | 'radio' description text, -- surfaced in the settings UI updated_by text, -- user_id of last editor (audit) updated_at timestamptz NOT NULL DEFAULT now(), - instance_id uuid NOT NULL REFERENCES instances(id) + instance_id uuid NOT NULL REFERENCES instances(id), + PRIMARY KEY (instance_id, key) -- per-tenant: one row per key PER instance (NOT key alone) ); ``` diff --git a/docs/plans/next-gen/components/auth.md b/docs/plans/next-gen/components/auth.md index 0114900..dcf99e2 100644 --- a/docs/plans/next-gen/components/auth.md +++ b/docs/plans/next-gen/components/auth.md @@ -214,7 +214,14 @@ fastify.addHook("preHandler", async (request: FastifyRequest, reply: FastifyRepl }); ``` -**`GET/POST /setup`** — a multi-step wizard (served server-rendered so it works before the SPA bootstrap): +> **Prefer a SPA route over SSR (F12).** Server-rendering the wizard reintroduces the Jinja-style HTML +> shell that Principle 5 (static, CDN-cacheable shell) exists to eliminate. The preferred implementation is +> a **normal SPA route** (`/setup`) gated by a `needs_setup` boolean in the public `/api/v1/config` +> payload: the static shell loads, sees `needs_setup = true`, and renders the wizard client-side — no +> server-rendered HTML, no second templating path in the web tier. The `POST /setup` endpoints stay as a +> plain JSON API. Server-rendering is a fallback only if the shell must not ship at all pre-setup. + +**`GET/POST /setup`** — a multi-step wizard (rendered by the SPA when `config.needs_setup` is true; see the note above): 1. **Welcome / network identity** — network name, city, country, contact (writes the Tier-2 branding settings that are otherwise empty on a fresh DB). 2. **Admin account** — username + password + confirm. Creates the first admin (see bootstrap insert sequence below). diff --git a/docs/plans/next-gen/components/data-model.md b/docs/plans/next-gen/components/data-model.md index 42c8a10..49a92d7 100644 --- a/docs/plans/next-gen/components/data-model.md +++ b/docs/plans/next-gen/components/data-model.md @@ -51,14 +51,16 @@ Route matching is **subsequence logic** over packet paths, not pure time-bucketi - `routes`, `route_nodes`, `route_observers` stay (definitions) — add the missing `UNIQUE(route_id, position)` and `UNIQUE(route_id, node_id)` constraints. - `route_results`, `route_result_history`, `route_recent_matches` stay as **worker-maintained tables** — but maintained by the single `DerivedStateWorker`, not 2 inline background cadences. The matcher reads `raw_receptions.path_hashes` directly (D5 outcome). -- Continuous aggregates are reserved for the **dashboard** time-bucketing (daily message/advert/packet counts, breakdowns, node-count history). +- Continuous aggregates are reserved for the **dashboard** time-bucketing that sources from a hypertable — daily **packet** counts and the packet breakdown by type (both over `raw_receptions`). Daily message/advert counts and node-count history source from OLTP/entity tables, so they are **worker-maintained rollup tables** (§3.6a), not CAGGs — the same rule as route health. -Net: the route-health subsystem loses 2 background threads + the inline maintenance + the write-amplified hops table, but keeps its 3 derived tables. The bigger CAGG win is on the dashboard. +Net: the route-health subsystem loses 2 background threads + the inline maintenance + the write-amplified hops table, but keeps its 3 derived tables. The CAGG win is the packet counts/breakdown; the dashboard's dedup'd-event counts move to cheap worker-maintained rollups. ### 1.5 Roles & security (S3, S4) - `user_profile_roles` join table (one row per profile × role) instead of CSV text. Indexable, constrainable. - **Row-level `instance_id` column** on every tenant-scoped table + Postgres RLS policies, **in addition to** `search_path`. Defense in depth: even a leaking connection cannot cross instances. (D3 locks row-level `instance_id` + RLS; schema-per-instance stays as an optional belt-and-braces layer.) +- **RLS must be forced and the app must not own the tables.** Postgres skips RLS for a table's owner. The migration/DDL role owns the tables; the application connects as a **separate non-owner role** (`meshcore_app`), and every table sets `FORCE ROW LEVEL SECURITY` so even a mistakenly-privileged connection is still policy-checked. Without both of these, RLS is silently inert. +- **All uniqueness is instance-scoped from Phase 0.** Every "unique" business key is `UNIQUE (instance_id, …)` — `nodes.public_key`, the dedup'd-event `event_hash` columns, `channels.name`/`key_hex`, `settings (instance_id, key)`. A physical node or a repeated on-air packet legitimately appears once *per tenant*; a global unique would let one tenant's row block another's insert. In single-tenant mode there is one instance, so this is behaviourally identical — but it is what makes Phase 7 genuinely additive (D21). ### 1.6 Naming consistency @@ -94,9 +96,11 @@ HYPERTABLES (TimescaleDB, compressed) telemetry(received_at, id, node_id, parsed_data jsonb, object_key, ...) event_logs(received_at, id, event_type, payload jsonb, ...) -- D2 locked -CONTINUOUS AGGREGATES (dashboard time-bucketing — the A7 win) - cagg_daily_message_counts, cagg_daily_advert_counts, cagg_daily_packet_counts, - cagg_packet_breakdown_by_type, cagg_node_count_history +CONTINUOUS AGGREGATES (only over the raw_receptions hypertable — the A7 win) + cagg_daily_packet_counts, cagg_packet_breakdown_by_type + +DASHBOARD ROLLUPS (worker-maintained — sources aren't hypertables, so NOT CAGGs) + dashboard_daily_message_counts, dashboard_daily_advert_counts, dashboard_node_count_history ROUTE HEALTH (worker-maintained — NOT CAGGs; subsequence logic) route_results(route_id PK, state, quality, quality_avg, matched_count, ..., instance_id) @@ -139,18 +143,29 @@ A single-row seed (`INSERT INTO instances ...`) is created by the initial migrat ```sql ALTER TABLE ENABLE ROW LEVEL SECURITY; +ALTER TABLE FORCE ROW LEVEL SECURITY; -- also enforce for the table owner CREATE POLICY tenant_isolation ON - USING (instance_id = current_setting('app.instance_id', true)::uuid); + USING (instance_id = current_setting('app.instance_id', true)::uuid) + WITH CHECK (instance_id = current_setting('app.instance_id', true)::uuid); -- block cross-instance writes too ``` -The app issues `SET LOCAL app.instance_id = '';` at the start of each transaction (node-postgres pool hook + Drizzle's `transaction()` callback). This is defense-in-depth on top of the existing schema-per-instance option. +Roles: the DDL/migration role owns the tables; the application connects as a **non-owner role** +(`meshcore_app`, granted DML only). Owner-bypass is why `FORCE` is required. + +The app issues `SET LOCAL app.instance_id = '';` at the start of **every transaction — reads +included**. Because `SET LOCAL` is transaction-scoped, any statement run in autocommit sees a NULL GUC +and the policy returns 0 rows. The API therefore wraps **each request** in a transaction (a Fastify +`preHandler`/`onRequest` hook opens the tx and issues the `SET LOCAL` before the handler runs); the +`DerivedStateWorker` and `IngestWorker` already do this per job/batch. This is the one RLS footgun to +guard in review — a read path that bypasses the request transaction silently loses tenant scoping. +This is defense-in-depth on top of the existing schema-per-instance option. ### 3.3 Entities (OLTP) ```sql CREATE TABLE nodes ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), - public_key char(64) UNIQUE NOT NULL, + public_key char(64) NOT NULL, name text, adv_type text, flags int, @@ -161,7 +176,8 @@ CREATE TABLE nodes ( last_seen timestamptz, instance_id uuid NOT NULL REFERENCES instances(id), created_at timestamptz NOT NULL DEFAULT now(), - updated_at timestamptz NOT NULL DEFAULT now() + updated_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (instance_id, public_key) -- per-tenant: the same physical node exists once per instance ); CREATE TABLE node_tags ( @@ -201,19 +217,21 @@ CREATE TABLE user_profile_nodes ( node_id uuid NOT NULL REFERENCES nodes(id) ON DELETE CASCADE, adopted_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (user_profile_id, node_id), - UNIQUE (node_id) -- one adopter per node (RLS scopes per instance) + UNIQUE (node_id) -- one adopter per node; node_id is already instance-specific (per-tenant node rows) ); CREATE TABLE channels ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), - name text UNIQUE NOT NULL, - key_hex text UNIQUE NOT NULL, + name text NOT NULL, + key_hex text NOT NULL, key_hash smallint NOT NULL, -- first byte of sha256(key); API exposes as 2-hex visibility channel_visibility NOT NULL DEFAULT 'community', enabled boolean NOT NULL DEFAULT true, instance_id uuid NOT NULL REFERENCES instances(id), created_at timestamptz NOT NULL DEFAULT now(), - updated_at timestamptz NOT NULL DEFAULT now() + updated_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (instance_id, name), + UNIQUE (instance_id, key_hex) ); ``` @@ -263,7 +281,7 @@ CREATE TABLE route_observers ( ```sql CREATE TABLE messages ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), - event_hash bytea UNIQUE NOT NULL, -- sha256(content) truncated to 16 bytes + event_hash bytea NOT NULL, -- sha256(content) truncated to 16 bytes kind message_kind NOT NULL, -- 'contact' | 'channel' pubkey_prefix char(12), channel_idx int, @@ -279,7 +297,8 @@ CREATE TABLE messages ( spam_score real, received_at timestamptz NOT NULL DEFAULT now(), instance_id uuid NOT NULL REFERENCES instances(id), - created_at timestamptz NOT NULL DEFAULT now() + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (instance_id, event_hash) -- dedup is per-tenant (multi-tenancy.md §10) ); CREATE INDEX ix_messages_kind_received ON messages(kind, received_at); CREATE INDEX ix_messages_prefix_received ON messages(pubkey_prefix, received_at); @@ -289,7 +308,7 @@ CREATE INDEX ix_messages_pathprefix_received ON messages(path_prefix, received_a CREATE TABLE advertisements ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), - event_hash bytea UNIQUE NOT NULL, + event_hash bytea NOT NULL, public_key char(64) NOT NULL, name text, adv_type text, @@ -298,7 +317,8 @@ CREATE TABLE advertisements ( advert_timestamp timestamptz, received_at timestamptz NOT NULL DEFAULT now(), instance_id uuid NOT NULL REFERENCES instances(id), - created_at timestamptz NOT NULL DEFAULT now() + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (instance_id, event_hash) -- per-tenant dedup ); CREATE INDEX ix_adverts_pubkey ON advertisements(public_key); CREATE INDEX ix_adverts_route_type_received ON advertisements(route_type, received_at); -- previously unindexed @@ -306,7 +326,7 @@ CREATE INDEX ix_adverts_received ON advertisements(received_at); CREATE TABLE trace_paths ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), - event_hash bytea UNIQUE NOT NULL, + event_hash bytea NOT NULL, initiator_tag bigint NOT NULL, path_len int, flags int, @@ -315,7 +335,8 @@ CREATE TABLE trace_paths ( hop_count int, received_at timestamptz NOT NULL DEFAULT now(), instance_id uuid NOT NULL REFERENCES instances(id), - created_at timestamptz NOT NULL DEFAULT now() + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (instance_id, event_hash) -- per-tenant dedup ); CREATE INDEX ix_trace_initiator ON trace_paths(initiator_tag); CREATE INDEX ix_trace_received ON trace_paths(received_at); @@ -330,7 +351,7 @@ TimescaleDB. Note: hypertables require the time column in the PK. CREATE TABLE raw_receptions ( received_at timestamptz NOT NULL, id bigint GENERATED ALWAYS AS IDENTITY, -- cheap, hypertable-friendly - observer_node_id uuid REFERENCES nodes(id) ON DELETE SET NULL, + observer_node_id uuid, -- loose ref to nodes.id; NO FK (see note below) wire_hash char(32), -- LetsMesh on-air hash; Nats-Msg-Id source event_hash bytea, -- backlink to dedup'd event, filled post-dispatch instance_id uuid NOT NULL REFERENCES instances(id), @@ -361,13 +382,18 @@ CREATE INDEX ix_raw_pathhash_gin ON raw_receptions USING gin(path_hashe ALTER TABLE raw_receptions SET (timescaledb.compress, timescaledb.compress_segmentby = 'observer_node_id'); SELECT add_compression_policy('raw_receptions', INTERVAL '24 hours'); SELECT add_retention_policy('raw_receptions', INTERVAL '30 days'); +-- NOTE (F6): the four hypertables reference nodes.id as a LOOSE uuid (no FK, no ON DELETE action). +-- A real FK with ON DELETE SET NULL/CASCADE would force cross-chunk DML — including on COMPRESSED +-- chunks, where DML is restricted/expensive — every time hourly node cleanup deletes a node. A dangling +-- node id is harmless here (same tolerance the plan already accepts for orphaned event_observers rows): +-- history repopulates and ages out on retention. RLS still scopes rows by instance_id. -- event_observers (junction, also hypertable) CREATE TABLE event_observers ( observed_at timestamptz NOT NULL, event_hash bytea NOT NULL, event_type text NOT NULL, - observer_node_id uuid NOT NULL REFERENCES nodes(id) ON DELETE CASCADE, + observer_node_id uuid NOT NULL, -- loose ref to nodes.id; NO FK (hypertable, see F6 note) snr real, path_len int, instance_id uuid NOT NULL REFERENCES instances(id), @@ -387,12 +413,15 @@ SELECT add_retention_policy('event_observers', INTERVAL '30 days'); CREATE TABLE telemetry ( received_at timestamptz NOT NULL, id uuid DEFAULT gen_random_uuid(), - node_id uuid REFERENCES nodes(id) ON DELETE SET NULL, + node_id uuid, -- loose ref to nodes.id; NO FK (hypertable, see F6 note) node_public_key char(64) NOT NULL, parsed_data jsonb, object_key text, - event_hash bytea, -- backlink to dedup'd event; NOT UNIQUE (hypertable constraint — - -- dedup handled by NATS Nats-Msg-Id + worker-side existence check) + event_hash bytea, -- backlink to dedup'd event; NOT UNIQUE (hypertable constraint). + -- Dedup is BEST-EFFORT: Nats-Msg-Id window dedup suppresses + -- redelivery, but two worker replicas can still both insert a + -- telemetry row (no unique to catch the race). Residual + -- duplicates are de-duplicated on read by (instance_id, event_hash). instance_id uuid NOT NULL REFERENCES instances(id), PRIMARY KEY (received_at, id) ); @@ -400,12 +429,15 @@ SELECT create_hypertable('telemetry', 'received_at', chunk_time_interval => INTE CREATE INDEX ix_telemetry_node_received ON telemetry(node_id, received_at); CREATE INDEX ix_telemetry_parsed_gin ON telemetry USING gin(parsed_data); SELECT add_compression_policy('telemetry', INTERVAL '24 hours'); +-- Optional retention (telemetry is otherwise unbounded). Enabled when tuning.telemetry_retention_days +-- is set; the retention job (derived-state.md) keeps the policy in sync with the setting. +-- SELECT add_retention_policy('telemetry', INTERVAL '90 days'); -- event_logs (renamed from events_log for naming consistency; D2 locked) CREATE TABLE event_logs ( received_at timestamptz NOT NULL, id uuid DEFAULT gen_random_uuid(), - observer_node_id uuid REFERENCES nodes(id) ON DELETE SET NULL, + observer_node_id uuid, -- loose ref to nodes.id; NO FK (hypertable, see F6 note) event_type text NOT NULL, payload jsonb, instance_id uuid NOT NULL REFERENCES instances(id), @@ -416,22 +448,89 @@ SELECT add_compression_policy('event_logs', INTERVAL '24 hours'); SELECT add_retention_policy('event_logs', INTERVAL '30 days'); ``` -Continuous aggregates (the dashboard win — replaces fan-out COUNTs): +Continuous aggregates (the dashboard win — replaces fan-out COUNTs). + +**A continuous aggregate can only be built over a hypertable.** `messages` and `advertisements` are +deliberately plain OLTP tables (content-hash dedup needs a global `event_hash` unique that a hypertable +can't provide). So **only the two `raw_receptions`-sourced counts are true CAGGs**; the dedup'd-event and +node-count rollups the dashboard needs are **worker-maintained tables** (see §3.7a) — the same +"can't be a CAGG, so the worker owns it" rule the route-health tables follow (§1.4). ```sql -CREATE MATERIALIZED VIEW cagg_daily_message_counts +-- Valid CAGGs (source = raw_receptions, a hypertable): +CREATE MATERIALIZED VIEW cagg_daily_packet_counts WITH (timescaledb.continuous) AS - SELECT date_trunc('day', received_at) AS day, kind, channel_idx, - count(*) AS cnt, instance_id - FROM messages GROUP BY 1, 2, 3, 4 + SELECT date_trunc('day', received_at) AS day, instance_id, count(*) AS cnt + FROM raw_receptions GROUP BY 1, 2 WITH NO DATA; -SELECT add_continuous_aggregate_policy('cagg_daily_message_counts', +SELECT add_continuous_aggregate_policy('cagg_daily_packet_counts', + start_offset => INTERVAL '7 days', end_offset => INTERVAL '1 hour', + schedule_interval => INTERVAL '5 minutes'); + +CREATE MATERIALIZED VIEW cagg_packet_breakdown_by_type +WITH (timescaledb.continuous) AS + SELECT date_trunc('day', received_at) AS day, instance_id, event_type, count(*) AS cnt + FROM raw_receptions GROUP BY 1, 2, 3 + WITH NO DATA; +SELECT add_continuous_aggregate_policy('cagg_packet_breakdown_by_type', start_offset => INTERVAL '7 days', end_offset => INTERVAL '1 hour', schedule_interval => INTERVAL '5 minutes'); --- Similarly: cagg_daily_advert_counts, cagg_daily_packet_counts (over raw_receptions), --- cagg_packet_breakdown_by_type, cagg_node_count_history. ``` +**RLS note:** RLS policies on the underlying hypertable do **not** propagate to a continuous aggregate. +Dashboard reads of these CAGGs must therefore carry an explicit `WHERE instance_id = current_setting( +'app.instance_id')::uuid` predicate (the API adds it from the `Principal`) — the CAGG is not a +tenant-safe surface on its own. + +### 3.6a Dashboard rollups (worker-maintained — the counts that can't be CAGGs) + +Daily **message** counts, **advert** counts, and **node-count history** are sourced from OLTP tables +(`messages`, `advertisements`) or from entity state (`nodes`) that has no append-only time column — none +can be a continuous aggregate. They are refreshed by the `dashboard-rollups` DerivedStateWorker job +(see [derived-state.md](derived-state.md#job-manifest)) as instance-scoped, RLS'd tables: + +```sql +CREATE TABLE dashboard_daily_message_counts ( + day date NOT NULL, + kind message_kind NOT NULL, + channel_idx int, + cnt int NOT NULL, + instance_id uuid NOT NULL REFERENCES instances(id), + PRIMARY KEY (instance_id, day, kind, channel_idx) +); +ALTER TABLE dashboard_daily_message_counts ENABLE ROW LEVEL SECURITY; +ALTER TABLE dashboard_daily_message_counts FORCE ROW LEVEL SECURITY; +CREATE POLICY tenant_isolation ON dashboard_daily_message_counts + USING (instance_id = current_setting('app.instance_id', true)::uuid); + +CREATE TABLE dashboard_daily_advert_counts ( + day date NOT NULL, + route_type text, + cnt int NOT NULL, + instance_id uuid NOT NULL REFERENCES instances(id), + PRIMARY KEY (instance_id, day, route_type) +); +ALTER TABLE dashboard_daily_advert_counts ENABLE ROW LEVEL SECURITY; +ALTER TABLE dashboard_daily_advert_counts FORCE ROW LEVEL SECURITY; +CREATE POLICY tenant_isolation ON dashboard_daily_advert_counts + USING (instance_id = current_setting('app.instance_id', true)::uuid); + +CREATE TABLE dashboard_node_count_history ( + day date NOT NULL, + active_nodes int NOT NULL, -- nodes with last_seen on `day` + total_nodes int NOT NULL, -- cumulative known nodes as of `day` + instance_id uuid NOT NULL REFERENCES instances(id), + PRIMARY KEY (instance_id, day) +); +ALTER TABLE dashboard_node_count_history ENABLE ROW LEVEL SECURITY; +ALTER TABLE dashboard_node_count_history FORCE ROW LEVEL SECURITY; +CREATE POLICY tenant_isolation ON dashboard_node_count_history + USING (instance_id = current_setting('app.instance_id', true)::uuid); +``` + +The job upserts completed-day buckets (`INSERT … ON CONFLICT (…) DO UPDATE`) each run — idempotent, +cheap (a handful of `GROUP BY` queries), and RLS-scoped like every other tenant table. + ### 3.7 Route health (worker-maintained, not CAGGs) Route matching is subsequence logic — it cannot be a CAGG. Three derived tables are maintained by the `route-evaluator` (300s) and `route-history` (3600s) jobs in the [DerivedStateWorker](derived-state.md#route-quality-averaging-route-history-job). `quality` is the current snapshot; `quality_avg` is the rolling 7-day average that the frontend prefers (see the [quality averaging algorithm](derived-state.md#route-quality-averaging-route-history-job) for the ordinal mapping and thresholds). @@ -468,7 +567,9 @@ CREATE POLICY tenant_isolation ON route_result_history CREATE TABLE route_recent_matches ( route_id uuid NOT NULL REFERENCES routes(id) ON DELETE CASCADE, - raw_reception_rowid bigint NOT NULL, -- references raw_receptions.id (loose; hypertable FK) + raw_reception_rowid bigint NOT NULL, -- references raw_receptions.id (loose; hypertable, no FK) + raw_reception_received_at timestamptz NOT NULL, -- store the partition key so match lookups get chunk + -- exclusion (raw_receptions PK is (received_at, id)) first_position int NOT NULL, last_position int NOT NULL, instance_id uuid NOT NULL REFERENCES instances(id), -- denormalized for RLS @@ -503,13 +604,14 @@ CREATE TABLE local_users ( -- Runtime settings (D11 — see api.md) CREATE TABLE settings ( - key text PRIMARY KEY, + key text NOT NULL, value jsonb NOT NULL, category text NOT NULL, -- 'branding' | 'features' | 'tuning' | 'webhooks' | 'radio' description text, updated_by text, updated_at timestamptz NOT NULL DEFAULT now(), - instance_id uuid NOT NULL REFERENCES instances(id) + instance_id uuid NOT NULL REFERENCES instances(id), + PRIMARY KEY (instance_id, key) -- per-tenant settings (one row per key PER instance) ); -- Custom pages (D20 — moved from file-based CONTENT_HOME to DB) diff --git a/docs/plans/next-gen/components/derived-state.md b/docs/plans/next-gen/components/derived-state.md index c319ef3..ea9876b 100644 --- a/docs/plans/next-gen/components/derived-state.md +++ b/docs/plans/next-gen/components/derived-state.md @@ -10,7 +10,7 @@ One `DerivedStateWorker` process (or a sidecar mode of the collector) owns all periodic work. The wins: - One set of metrics, one shutdown path, one retry policy. -- **Dashboard aggregations become TimescaleDB continuous aggregates** (CAGGs). Daily message/advert/packet counts, packet-breakdown by event type, and node-count history are all *pure time-bucketing* — the textbook CAGG use case. The worker refreshes them every few minutes; the API reads precomputed buckets instead of issuing fan-out COUNTs (A7). This is where CAGGs pay off biggest. +- **Dashboard aggregations are precomputed** instead of fan-out COUNTs (A7). Two of them — daily **packet** counts and the **packet breakdown by type** — source from the `raw_receptions` hypertable and are true TimescaleDB **continuous aggregates**. The other three — daily **message** counts, **advert** counts, and **node-count history** — source from OLTP/entity tables (`messages`, `advertisements`, `nodes`) that are *not* hypertables, so a CAGG cannot be built over them; the `dashboard-rollups` job maintains them as plain rollup tables (data-model.md §3.6a). Either way the API reads precomputed buckets, not live COUNTs. - **Route health stays as 3 derived tables** (`route_results`, `route_result_history`, `route_recent_matches`) maintained by the worker — *not* CAGGs. Route matching is subsequence logic over `raw_receptions.path_hashes`, which cannot be expressed as a fixed time-bucket aggregate. The wins here are (a) the matcher reads one array column instead of a write-amplified hops table (D5), and (b) one worker maintains all three tables + the preview endpoint instead of 2 background cadences + inline maintenance. - Spam rescoring can become a **DB function** invoked on insert (for the online score) plus a periodic sweep (for the symmetric hindsight score), instead of a Python loop issuing per-row queries. - Retention becomes **chunked** (drop-old-chunks for hypertables via TimescaleDB retention policies) — no multi-second exclusive locks (W10). @@ -25,9 +25,15 @@ One process — `DerivedStateWorker` — owns every periodic job. Replaces the s | `route-history` | 3600s | `route-history-backfill` thread (3600s) | Refresh completed UTC-day buckets in `route_result_history` over the raw retention window, then recompute `route_results.quality_avg` (rolling 7-day average quality tier from the updated history + current snapshot). | Per (route, day): upsert keyed on the composite PK. | | `spam-rescore` | 120s | `spam-rescore` thread (120s) | Symmetric-window rescore of recent `messages.spam_score` (see Spam rescoring below). | Per message: only writes when the score changes. | | `retention` | hourly | `cleanup` thread (hourly) | Drop expired hypertable chunks (`raw_receptions`, `event_logs`, `event_observers` at 30d); `cleanup_inactive_nodes`; `recompute_observer_flags`. Chunked (see Chunked retention). | Chunk drops are idempotent; node cleanup is keyed on `last_seen`. | +| `dashboard-rollups` | 300s | (new — the counts that can't be CAGGs, F2) | Upsert completed-day buckets into `dashboard_daily_message_counts`, `dashboard_daily_advert_counts`, `dashboard_node_count_history` (data-model.md §3.6a). These source from OLTP/entity tables, so they cannot be TimescaleDB continuous aggregates. | Per (instance, day, …): `INSERT … ON CONFLICT DO UPDATE`. | | `metrics-gauges` | 60s | (new — replaces the metrics COUNT fan-out A7) | Precompute the Prometheus gauge values into a `_metrics_cache` table; `/metrics` reads the cache (TTL-cached today, but the computation moves here). | Overwrite. | | `cagg-health` | 300s | (new) | Assert each CAGG's `refresh_status` is recent; log + alert if stale. Read-only check. | — | +> **CAGGs vs rollups:** only `cagg_daily_packet_counts` and `cagg_packet_breakdown_by_type` (over the +> `raw_receptions` hypertable) are true continuous aggregates. Daily message/advert counts and node-count +> history come from OLTP/entity tables and are maintained by the `dashboard-rollups` job above — the same +> "can't be a CAGG, so the worker owns it" rule route health follows. + ## Scheduler implementation A small home-grown loop (no APScheduler dependency). Each job is a registered `PeriodicJob` with: name, interval, `async def run(session)`, and a `pg_advisory_lock` key. @@ -57,7 +63,12 @@ class DerivedStateWorker: async def _run_one(self, job: PeriodicJob) -> None: async with self.sessions() as s: - await s.execute(text("SELECT pg_advisory_xact_lock(:k)"), {"k": job.lock_key}) + # Two-arg advisory lock: (job key, stable per-instance key). Use hashtext(instance_id) — NOT a + # positional instance_index, which shifts as tenants come/go and can differ between replicas, + # letting the same (job, instance) run twice (F7). The two-arg form also avoids cross-job + # collisions that `base_key + index` risks. + await s.execute(text("SELECT pg_advisory_xact_lock(:j, hashtext(:iid))"), + {"j": job.lock_key, "iid": str(self.instance_id)}) await s.execute(text("SET LOCAL app.instance_id = :id"), {"id": self.instance_id}) try: await job.run(s) @@ -115,7 +126,22 @@ $$ LANGUAGE plpgsql STABLE; ``` - **At insert** (IngestWorker): `UPDATE messages SET spam_score = compute_spam_score(id, ...) WHERE id = ?;` -- **Sweep** (`spam-rescore` job): `UPDATE messages SET spam_score = compute_spam_score(id, ...) WHERE received_at > now() - '1 hour'::interval AND spam_score IS DISTINCT FROM compute_spam_score(id, ...);` — only writes changed rows. +- **Sweep** (`spam-rescore` job): compute the score **once per candidate** in a subquery, then filter and + write from that single value — do **not** call the function in both `WHERE` and `SET` (it is COUNT-heavy + and would run twice per row every 120s): + +```sql +UPDATE messages m +SET spam_score = s.new_score +FROM ( + SELECT id, compute_spam_score(id, :window, :min_path_hops, :path_threshold, + :name_threshold, :w_path, :w_name) AS new_score + FROM messages + WHERE received_at > now() - INTERVAL '1 hour' +) s +WHERE m.id = s.id + AND m.spam_score IS DISTINCT FROM s.new_score; -- only rows whose score actually changed +``` This kills the asymmetric-online / symmetric-sweep split's Python implementation (`spam.py` ~315 LOC collapses to the function + two call sites) while preserving the documented behaviour. @@ -219,6 +245,14 @@ async def retention_job(session: AsyncSession) -> None: `cleanup_inactive_nodes` and `recompute_observer_flags` follow the same chunked pattern. +**Node cleanup does not touch the hypertables (F6).** The hypertable columns that point at `nodes` +(`raw_receptions.observer_node_id`, `event_observers.observer_node_id`, `telemetry.node_id`, +`event_logs.observer_node_id`) are **loose `uuid` references with no FK** (data-model.md §3.6). If they were +real FKs with `ON DELETE SET NULL/CASCADE`, deleting an inactive node would force DML across hypertable +chunks — including compressed ones, where DML is restricted/costly — turning an hourly cleanup into a +decompress storm. Instead, a deleted node simply leaves a dangling id in history (harmless, like the +orphaned `event_observers` rows above); it compresses and ages out on retention. + ## Observability Each job emits: diff --git a/docs/plans/next-gen/components/infrastructure.md b/docs/plans/next-gen/components/infrastructure.md index 33fd575..217e283 100644 --- a/docs/plans/next-gen/components/infrastructure.md +++ b/docs/plans/next-gen/components/infrastructure.md @@ -106,7 +106,7 @@ Existing operators migrate via the `db migrate-to-postgres` runbook before upgra NATS JetStream owns two roles in the target architecture: -1. **Durable ingest stream** (`meshcore.ingest.`) — the MqttIngester produces decoded envelopes; the `IngestWorker` consumer group reads them. JetStream gives at-least-once delivery, disk persistence, and **server-side dedup** via the `Nats-Msg-Id` header (set to the packet's `wire_hash`) within a configurable duplicate window — so MQTT redelivery does not double-process. +1. **Durable ingest stream** — a single platform-wide `INGEST` stream capturing `meshcore.ingest.>` (per-instance subject tokens `meshcore.ingest..`). The MqttIngester produces decoded envelopes; the shared `IngestWorker` consumer group reads them. One stream (not one-per-instance) so the Phase 7 wildcard consumer group works — see [D4](../decisions/D04-nats-jetstream-ingest.md) (F8). JetStream gives at-least-once delivery, disk persistence, and **server-side dedup** via the `Nats-Msg-Id` header (packet `wire_hash`) within a configurable duplicate window — so MQTT redelivery does not double-process. 2. **Realtime fan-out bus** (`events.new.`) — workers publish a small "new event" notification after commit; the API's SSE endpoint subscribes and pushes to clients. ### Why NATS over the alternatives @@ -120,9 +120,9 @@ NATS JetStream owns two roles in the target architecture: ### Stream configuration -The default Compose stack gains a `nats` service with a JetStream persistence volume. Operators of the existing bundled-Redis cache keep Redis; the ingest path no longer touches it. +The default Compose stack gains a `nats` service with a JetStream persistence volume. Operators of the existing bundled-Redis cache keep Redis; the ingest path no longer touches it. One `INGEST` stream is created at provisioning (subject `meshcore.ingest.>`) with one durable consumer `workers` that all IngestWorker replicas bind to. -- `duplicate_window = 5m` — server-side dedup on `Nats-Msg-Id` = packet `wire_hash`. +- `duplicate_window = 5m` — server-side dedup on `Nats-Msg-Id` = packet `wire_hash` (tenant-prefixed in multi-tenant mode). - `max_age = 7d` — replay window for worker restarts. - `storage = file`. - `retention = limits`. diff --git a/docs/plans/next-gen/components/ingest.md b/docs/plans/next-gen/components/ingest.md index 8834065..2068306 100644 --- a/docs/plans/next-gen/components/ingest.md +++ b/docs/plans/next-gen/components/ingest.md @@ -69,7 +69,7 @@ def persist_deduped_event( ``` - Computes the SHA-256 content hash. -- `INSERT ... ON CONFLICT (event_hash) DO NOTHING` (Postgres-native; drop the dialect branch). +- `INSERT ... ON CONFLICT (instance_id, event_hash) DO NOTHING` (Postgres-native; drop the dialect branch). The conflict target is the **composite** `(instance_id, event_hash)` unique — dedup is per-tenant, so the same physical packet creates one event *per* instance (multi-tenancy.md §10). - On conflict, just attaches the observer. - Returns whether it was a new event (drives the pub/sub fan-out — only fire webhooks/realtime on first sighting). @@ -86,8 +86,9 @@ async def persist_deduped_event( observer_id: UUID, observer_meta: ObserverMeta, ) -> DedupResult: - """INSERT ... ON CONFLICT (event_hash) DO NOTHING; attach observer either way. - Returns DedupResult(is_new, event_id, event_hash). Native Postgres — no dialect branch.""" + """INSERT ... ON CONFLICT (instance_id, event_hash) DO NOTHING; attach observer either way. + Composite conflict target = per-tenant dedup (F1). Returns DedupResult(is_new, event_id, event_hash). + Native Postgres — no dialect branch.""" ... # Each handler is ~15 lines, not ~50. @@ -177,15 +178,23 @@ Key insight: `decoded` stays on the row even when D8 is activated. Only `raw_hex ## 6. NATS topology -Two JetStream streams + one core (non-durable) subject: +One JetStream stream + two core (non-durable) subject families. The stream is **single and +platform-wide** — tenancy is a subject token, not a separate stream. This matters: a JetStream consumer +group cannot span multiple streams, and Phase 7's shared worker pool subscribes across all tenants with a +wildcard. One stream from Phase 0 makes multi-tenancy purely additive (D21). | Stream/Subject | Type | Subjects | Producers | Consumers | |---|---|---|---|---| -| `INGEST-` | JetStream, durable, `WorkQueuePolicy` | `meshcore.ingest..*` | MqttIngester | IngestWorker consumer group (`workers`, `ack_explicit`) | -| `events.new.` | Core pub/sub (non-durable) | `events.new..{messages,advertisements,...}` | IngestWorker (after commit) | API SSE endpoint(s) | +| `INGEST` | JetStream, durable, `WorkQueuePolicy` | `meshcore.ingest.>` (per-instance tokens: `meshcore.ingest..`) | MqttIngester | IngestWorker consumer group (durable `workers`, `ack_explicit`) — one shared consumer, all replicas bind to it | +| `events.new.` | Core pub/sub (non-durable) | `events.new..{messages,advertisements,...}` | IngestWorker (after commit) | API SSE endpoint(s), WebhookWorker | | `channel.keys.` | Core pub/sub | `channel.keys..updated` | IngestWorker (after channel mutation) | MqttIngester (reload `ChannelKeyCache`) | -Stream config: `duplicate_window = 5m` (server-side dedup on `Nats-Msg-Id` = packet `wire_hash`), `max_age = 7d` (replay window for worker restarts), `storage = file`, `retention = limits`. +Stream config: `duplicate_window = 5m` (server-side dedup on `Nats-Msg-Id` = packet `wire_hash`; in +multi-tenant mode the id is tenant-prefixed — multi-tenancy.md §4), `max_age = 7d` (replay window for +worker restarts), `storage = file`, `retention = limits`. + +> Single-tenant mode is just one instance's subjects flowing through the same `INGEST` stream — no +> per-instance stream to create, and the wildcard consumer already covers every future tenant. --- @@ -278,7 +287,9 @@ class IngestWorker: ) -> None: ... async run(): Promise { - const sub = await this.js.pullSubscribe("meshcore.ingest.*", { durable: "workers" }); + // Subscribe to the whole ingest subject tree (all instances). `meshcore.ingest.*` would only + // match a 3-token subject; the real subjects are 4-token (meshcore.ingest..). + const sub = await this.js.pullSubscribe("meshcore.ingest.>", { durable: "workers" }); while (this._running) { const msgs = await sub.fetch(this.batchSize, { timeout: 5000 }); await this._processBatch(msgs); @@ -347,7 +358,7 @@ async function touchNode(tx: Tx, env: IngestEnvelope, instanceId: string): Promi last_seen: env.ingested_at, instance_id: instanceId, }).onConflictDoUpdate({ - target: nodes.public_key, + target: [nodes.instance_id, nodes.public_key], // composite unique — per-tenant node rows set: { name: sql`COALESCE(EXCLUDED.name, nodes.name)`, adv_type: sql`EXCLUDED.adv_type`, @@ -368,13 +379,23 @@ async function touchNode(tx: Tx, env: IngestEnvelope, instanceId: string): Promi last_seen: env.ingested_at, instance_id: instanceId, }).onConflictDoUpdate({ - target: nodes.public_key, + target: [nodes.instance_id, nodes.public_key], set: { last_seen: sql`EXCLUDED.last_seen`, updated_at: sql`now()` }, }).returning(); return node.id; } ``` +### The observing node must exist too + +The reception/junction rows carry `observer_node_id` (a loose `nodes.id` reference — no FK, F6). The +worker therefore **find-or-creates the observing node** as well, using the same instance-scoped upsert as +`touchNode` (keyed on `(instance_id, observer.public_key)`), before writing `raw_receptions` / +`event_observers`. Because those columns are loose references (no FK), a stale id would not error — but +resolving it here keeps the observer's `nodes` row present and `last_seen` current. The observer upsert +and the source `touchNode` run in the same batch transaction (one round-trip each, deduplicated within +the batch when many envelopes share an observer). + - **Adverts** are the authoritative source for node metadata (name, type, flags, GPS). `COALESCE` preserves existing values when the advert field is null (partial adverts). - **Non-advert events** (messages, traces, telemetry) only bump `last_seen` — they don't carry name/GPS metadata. - **Observer flag** (`is_observer`): set from the envelope's `observer.is_observer` (derived from the MQTT topic's IATA code matching a known observer list). The `recompute_observer_flags` DerivedStateWorker job (daily) cross-checks against `nodes` that have recent observer-traffic but lack the flag. diff --git a/docs/plans/next-gen/components/migration.md b/docs/plans/next-gen/components/migration.md index c3d97bb..dbe5875 100644 --- a/docs/plans/next-gen/components/migration.md +++ b/docs/plans/next-gen/components/migration.md @@ -124,12 +124,18 @@ meshcore-hub admin diff-stacks \ --api-key ``` +> **Match on `wire_hash`, not `event_hash`.** The two stacks compute the content dedup hash with +> different algorithms (old = MD5, new = SHA-256 truncated), so the *same* event has a different +> `event_hash` in each stack — a coverage check keyed on `event_hash` would always report 0%. The +> LetsMesh on-air `wire_hash` is identical in both stacks, so it is the correct join key for verifying the +> new pipeline decoded the same packets. + **What it compares (per hour bucket, per event type):** | Check | Query | Pass condition | |---|---|---| | **Event count parity** | `GET /api/v1/messages?since=&until=` (and adverts, packets) — compare `total` | Counts match within ±2 (tolerance for race at hour boundaries) | -| **Hash coverage** | Sample 100 `event_hash` values from the old stack's hour; verify each exists in the new stack (`GET /api/v1/messages?event_hash=`) | 100% coverage | +| **Hash coverage** | Sample 100 `wire_hash` values from the old stack's hour; verify each exists in the new stack (`GET /api/v1/packet-groups?wire_hash=`) | 100% coverage | | **Observer parity** | For the sampled events, compare observer counts (`event_observers` junction) | Counts match exactly | | **Node count** | `GET /api/v1/nodes` — compare `total` | Within ±5 (nodes appear/disappear on advert timing) | diff --git a/docs/plans/next-gen/components/multi-tenancy.md b/docs/plans/next-gen/components/multi-tenancy.md index d1d9fbc..04140b9 100644 --- a/docs/plans/next-gen/components/multi-tenancy.md +++ b/docs/plans/next-gen/components/multi-tenancy.md @@ -68,12 +68,16 @@ flowchart TB ## 2. What's already free -The single-tenant design (Phases 0–6) built multi-tenant foundations by default. These require **zero changes**: +The single-tenant design (Phases 0–6) built multi-tenant foundations by default. These require **zero +changes** — provided the Phase 0 schema is instance-scoped *including its uniqueness and stream topology* +(the corrections in [review-findings.md](../review-findings.md) F1/F8 fold these into the base schema, so +"the schema does not change" is now true rather than aspirational): | Capability | Why it's free | |---|---| -| Per-tenant data isolation | `instance_id` + RLS on every table (D3) | -| Per-tenant settings/branding | `settings` table is per-instance (D11) | +| Per-tenant data isolation | `instance_id` + RLS on every table (D3), enforced with `FORCE ROW LEVEL SECURITY` + a non-owner app role | +| Per-tenant uniqueness | every business key is `UNIQUE (instance_id, …)` from Phase 0 — `nodes.public_key`, `event_hash`, `channels.name/key_hex`, `settings (instance_id, key)` (F1) | +| Per-tenant settings/branding | `settings` table is per-instance, PK `(instance_id, key)` (D11) | | Per-tenant custom pages | `custom_pages` table is per-instance (D20) | | Per-tenant channels/routes/tags/profiles | All tables carry `instance_id` | | Per-tenant NATS subjects | Already namespaced: `meshcore.ingest..*`, `events.new..*` | @@ -88,6 +92,11 @@ The single-tenant assumption lives in exactly **three places** that Phase 7 modi 2. `AuthMiddleware.instance_id` (from env) — becomes hostname/JWT resolution. 3. OIDC config (Tier-1 env vars) — gains a per-instance DB path. +The **NATS ingest stream is already single and platform-wide** (`INGEST`, subject `meshcore.ingest.>`, +one durable `workers` consumer — ingest.md §6), so the shared worker pool's wildcard subscription needs +no new stream in Phase 7. Had the stream been per-instance (`INGEST-`), the wildcard consumer group +could not span it — hence the single-stream choice from Phase 0 (F8). + --- ## 3. Observer → tenant mapping @@ -547,7 +556,7 @@ Workers are **tenant-agnostic**. They discover active tenants dynamically and pr |---|---| | **MqttIngester** | Already shared — decodes all MQTT traffic, routes per tenant via `ObserverAllowlistCache` (§4). New tenants picked up on `instance.created` NATS notification (cache reload). | | **IngestWorker** (pool) | Subscribes to `meshcore.ingest.>` (**wildcard — all tenants**). Each envelope carries `instance_id`; the worker sets `SET LOCAL app.instance_id` per batch. Consumer group `workers` distributes across replicas. New tenants need zero config — the wildcard subscription already covers their subject. | -| **DerivedStateWorker** | Queries `SELECT id FROM instances WHERE deleted_at IS NULL` at startup + on `instance.created`/`instance.deleted` NATS notifications. For each active instance, runs the job manifest with `SET LOCAL app.instance_id`. Per-instance advisory lock keys (`lock_key = base_key + instance_index`) prevent one tenant's long-running job from blocking another's. Round-robin across instances. | +| **DerivedStateWorker** | Queries `SELECT id FROM instances WHERE deleted_at IS NULL` at startup + on `instance.created`/`instance.deleted` NATS notifications. For each active instance, runs the job manifest with `SET LOCAL app.instance_id`. Per-instance advisory locks use the two-arg form `pg_advisory_xact_lock(job_key, hashtext(instance_id))` — a **stable** per-(job, instance) key, not a positional `base_key + instance_index` (which shifts as tenants come/go and can double-execute across replicas — F7). Round-robin across instances. | | **WebhookWorker** | Subscribes to `events.new.>` (**wildcard — all tenants**). Each event carries `instance_id`; the worker loads that tenant's webhook settings from `SettingsCache`. New tenants need zero config. | ### New-tenant pickup (zero operator intervention) diff --git a/docs/plans/next-gen/decisions/D01-timescaledb-for-history.md b/docs/plans/next-gen/decisions/D01-timescaledb-for-history.md index 5c64805..72277f4 100644 --- a/docs/plans/next-gen/decisions/D01-timescaledb-for-history.md +++ b/docs/plans/next-gen/decisions/D01-timescaledb-for-history.md @@ -11,7 +11,7 @@ The §13-D1 decision had to land before the §16 schema could be drafted — eve ## Decision -**PostgreSQL 17 + TimescaleDB** (community edition, Apache-2.0). The high-volume append-only streams (`raw_receptions`, `event_observers`, `telemetry`, `event_logs`) become TimescaleDB hypertables partitioned by `received_at` (1-day chunks), with columnar compression policies after 24h and retention policies (default 30d, up from 2d). The five dashboard time-bucketing workloads become **continuous aggregates** (`cagg_daily_message_counts`, `cagg_daily_advert_counts`, `cagg_daily_packet_counts`, `cagg_packet_breakdown_by_type`, `cagg_node_count_history`) refreshed on a 5-minute policy — replacing the fan-out COUNTs. +**PostgreSQL 17 + TimescaleDB** (community edition, Apache-2.0). The high-volume append-only streams (`raw_receptions`, `event_observers`, `telemetry`, `event_logs`) become TimescaleDB hypertables partitioned by `received_at` (1-day chunks), with columnar compression policies after 24h and retention policies (default 30d, up from 2d). The **hypertable-sourced** dashboard time-bucketing workloads become **continuous aggregates** (`cagg_daily_packet_counts`, `cagg_packet_breakdown_by_type`, both over `raw_receptions`) refreshed on a 5-minute policy — replacing the fan-out COUNTs. The other three dashboard rollups (daily message counts, advert counts, node-count history) source from OLTP/entity tables (`messages`, `advertisements`, `nodes`), which are **not** hypertables — a continuous aggregate cannot be built over them, so they are worker-maintained rollup tables instead (F2; data-model.md §3.6a). Route-health derived tables stay worker-maintained (subsequence logic is not a pure time-bucket aggregate — see §6.3.4 and D5). diff --git a/docs/plans/next-gen/decisions/D03-row-level-tenancy-rls.md b/docs/plans/next-gen/decisions/D03-row-level-tenancy-rls.md index ce66aa6..dc9307e 100644 --- a/docs/plans/next-gen/decisions/D03-row-level-tenancy-rls.md +++ b/docs/plans/next-gen/decisions/D03-row-level-tenancy-rls.md @@ -13,11 +13,25 @@ Today multi-tenant isolation is **connection-level only**: `SET search_path = in ```sql ALTER TABLE ENABLE ROW LEVEL SECURITY; +ALTER TABLE FORCE ROW LEVEL SECURITY; -- enforce even for the table owner (F3) CREATE POLICY tenant_isolation ON - USING (instance_id = current_setting('app.instance_id', true)::uuid); + USING (instance_id = current_setting('app.instance_id', true)::uuid) + WITH CHECK (instance_id = current_setting('app.instance_id', true)::uuid); ``` -The app issues `SET LOCAL app.instance_id = '';` at the start of every transaction (connection-pool hook + async-session `before_commit` hook). Schema-per-instance remains available as a belt-and-braces layer for operators who want physical separation; it is no longer the *only* boundary. +The app issues `SET LOCAL app.instance_id = '';` at the start of every transaction — **reads +included** (a Fastify per-request `preHandler` opens the tx and sets the GUC; the workers do it per +batch/job). Because `SET LOCAL` is transaction-scoped, a statement run in autocommit sees a NULL GUC and +the policy returns 0 rows — so no DB access may happen outside the request/job transaction. + +**Two enforcement prerequisites (F3), or RLS is silently inert:** +1. `FORCE ROW LEVEL SECURITY` on every policy'd table — Postgres skips RLS for the *owner* otherwise. +2. The application connects as a **non-owner role** (`meshcore_app`, DML-only); the DDL/migration role + owns the tables. The Phase 6 "cross-instance query returns 0 rows" audit must run as the app role, not + the owner, or it proves nothing. + +Schema-per-instance remains available as a belt-and-braces layer for operators who want physical +separation; it is no longer the *only* boundary. ## Consequences diff --git a/docs/plans/next-gen/decisions/D04-nats-jetstream-ingest.md b/docs/plans/next-gen/decisions/D04-nats-jetstream-ingest.md index 9e739d2..ef85180 100644 --- a/docs/plans/next-gen/decisions/D04-nats-jetstream-ingest.md +++ b/docs/plans/next-gen/decisions/D04-nats-jetstream-ingest.md @@ -11,7 +11,9 @@ Today's ingest is a single-threaded MQTT callback with no backpressure (W1): one **NATS 2.10+ JetStream** owns both roles: -1. **Durable ingest stream** (`INGEST-`, JetStream, `WorkQueuePolicy`, subject `meshcore.ingest..*`). `MqttIngester` produces decoded envelopes; the `IngestWorker` consumer group (`durable="workers"`, `ack_explicit`) reads them. Server-side dedup via the `Nats-Msg-Id` header set to the packet's `wire_hash` within a `duplicate_window = 5m`. `max_age = 7d` (replay window for worker restarts), `storage = file`, `retention = limits`. +1. **Durable ingest stream** — a **single, platform-wide `INGEST` stream** (JetStream, `WorkQueuePolicy`, subject `meshcore.ingest.>`; per-instance tokens `meshcore.ingest..`). `MqttIngester` produces decoded envelopes; the `IngestWorker` consumer group (`durable="workers"`, `ack_explicit`) reads them. Server-side dedup via the `Nats-Msg-Id` header (packet `wire_hash`, tenant-prefixed in multi-tenant mode) within a `duplicate_window = 5m`. `max_age = 7d` (replay window for worker restarts), `storage = file`, `retention = limits`. + + > **One stream, not one-per-instance (F8).** A JetStream consumer group cannot span multiple streams, and Phase 7's shared worker pool subscribes to `meshcore.ingest.>` across all tenants. Defining a per-instance `INGEST-` stream would break that wildcard consumer and require creating a stream per tenant at registration. A single stream from Phase 0 keeps multi-tenancy purely additive; single-tenant is just one instance's subjects. 2. **Realtime fan-out bus** (`events.new..{table}`, core non-durable pub/sub). Workers publish a small "event persisted" notification after commit; the API's SSE endpoint subscribes and pushes to clients. Redis narrows to optional API response cache only — it is no longer on the ingest or fan-out paths and may be omitted entirely. diff --git a/docs/plans/next-gen/decisions/D05-fold-packet-path-hops.md b/docs/plans/next-gen/decisions/D05-fold-packet-path-hops.md index 9615a4c..47d4d30 100644 --- a/docs/plans/next-gen/decisions/D05-fold-packet-path-hops.md +++ b/docs/plans/next-gen/decisions/D05-fold-packet-path-hops.md @@ -1,7 +1,7 @@ # D05: Fold `packet_path_hops` Into `raw_receptions.path_hashes` (Research Spike) -- **Status:** Locked (spike — decision lands in Phase 2) -- **Iteration:** 2 +- **Status:** Locked (spike — decision lands in **Phase 0**, before the DDL is authored) +- **Iteration:** 2 (rescheduled to Phase 0 in iteration 8 — F5) ## Context @@ -15,7 +15,7 @@ - `sweep_ms(F) ≤ 1.5 × sweep_ms(S)`, AND - `p95_route_ms(F) ≤ 500ms`. -**Fallback if the gate fails:** keep a separate `packet_path_hops` hypertable (still better than today because TimescaleDB-partitioned) **but stop denormalizing** `packet_hash`, `received_at`, `observer_node_id` (reachable via the FK to `raw_receptions`). The `raw_receptions.path_hashes` column stays either way (it backs the packet-group detail view). Budget half a day; record results in `docs/plans/-d5-fold-benchmark/`. Run **before** the §16 schema is frozen. +**Fallback if the gate fails:** keep a separate `packet_path_hops` hypertable (still better than today because TimescaleDB-partitioned) **but stop denormalizing** `packet_hash`, `received_at`, `observer_node_id` (reachable from `raw_receptions`). The `raw_receptions.path_hashes` column stays either way (it backs the packet-group detail view). Budget half a day; record results in `docs/plans/-d5-fold-benchmark/`. **Run in Phase 0, before the DDL migration is authored** — the benchmark needs only a throwaway TimescaleDB, so it has no dependency on any later phase, and scheduling it in Phase 2 (as originally written) created a circular dependency with the Phase 0 schema freeze (F5). ## Consequences diff --git a/docs/plans/next-gen/decisions/D16-two-replica-worker-ha.md b/docs/plans/next-gen/decisions/D16-two-replica-worker-ha.md index 5cdb527..5e60c8e 100644 --- a/docs/plans/next-gen/decisions/D16-two-replica-worker-ha.md +++ b/docs/plans/next-gen/decisions/D16-two-replica-worker-ha.md @@ -13,7 +13,10 @@ The §19.1 manifest consolidates six daemon threads into one `DerivedStateWorker ```typescript await db.transaction(async (tx) => { - await tx.execute(sql`SELECT pg_advisory_xact_lock(${job.lockKey})`); + // Two-arg lock: (job key, stable per-instance key). In multi-tenant mode use + // hashtext(instanceId) as the second arg — NOT a positional instance_index, which shifts as + // tenants come/go and can double-execute across replicas (F7). + await tx.execute(sql`SELECT pg_advisory_xact_lock(${job.lockKey}, hashtext(${instanceId}))`); await tx.execute(sql`SET LOCAL app.instance_id = ${instanceId}`); try { await job.run(tx); diff --git a/docs/plans/next-gen/decisions/D21-multi-tenancy.md b/docs/plans/next-gen/decisions/D21-multi-tenancy.md index 0ae5041..8435623 100644 --- a/docs/plans/next-gen/decisions/D21-multi-tenancy.md +++ b/docs/plans/next-gen/decisions/D21-multi-tenancy.md @@ -5,7 +5,15 @@ ## Context -The schema is already instance-scoped: every tenant table carries `instance_id` with RLS (D3), NATS subjects are namespaced (`meshcore.ingest..*`), cache keys are scoped, and settings/pages/channels/routes/tags/profiles are all per-instance. The single-tenant assumption lives in exactly three places: the MqttIngester's constructor (`instance_id` is a process-level arg), the API middleware (`instance_id` from env), and OIDC config (Tier-1 env vars). +The schema is instance-scoped from Phase 0: every tenant table carries `instance_id` with RLS (D3), NATS subjects are namespaced (`meshcore.ingest..*`), cache keys are scoped, and settings/pages/channels/routes/tags/profiles are all per-instance. The single-tenant assumption lives in exactly three places: the MqttIngester's constructor (`instance_id` is a process-level arg), the API middleware (`instance_id` from env), and OIDC config (Tier-1 env vars). + +> **Correction (iteration 8, F1/F8).** "The schema does not change" only holds because the base schema is +> built multi-tenant-ready — which required four fixes folded into Phase 0: (1) every business-key unique +> is `UNIQUE (instance_id, …)` (`nodes.public_key`, the `event_hash` columns, `channels.name/key_hex`); +> (2) `settings` PK is `(instance_id, key)`, not `key` alone; (3) a **single** platform-wide `INGEST` +> NATS stream (`meshcore.ingest.>`), so the Phase 7 wildcard consumer needs no new stream; (4) RLS is +> `FORCE`d and the app runs as a non-owner role. With global uniques or a per-instance stream, Phase 7 +> would in fact require schema/topology migrations. See [review-findings.md](../review-findings.md). The question (iteration 7): can multiple MeshCore communities share one platform deployment — each with their own branding, pages, OIDC, and observer pool — while the MQTT backend accepts all observers and each tenant chooses which ones they want? diff --git a/docs/plans/next-gen/implementation-checklist.md b/docs/plans/next-gen/implementation-checklist.md index 36b39bd..b1fe404 100644 --- a/docs/plans/next-gen/implementation-checklist.md +++ b/docs/plans/next-gen/implementation-checklist.md @@ -16,12 +16,23 @@ ## Phase 0 — Foundations +### D5 benchmark (before the schema is authored — F5) +- [ ] Write + run `bench/route_match_benchmark.ts` at Low/Medium/High shapes — [testing.md → D5 plan](testing.md#d5-benchmark-plan-fold-vs-separate) +- [ ] Record the fold-vs-separate decision (fold if `sweep_ms ≤ 1.5×` + `p95 ≤ 500ms` at High); the DDL below reflects the outcome + ### Schema -- [ ] Write the initial Drizzle Kit migration for the full [target DDL](components/data-model.md#3-phase-0--schema-ddl-target-authoritative) (enums, entities, hypertables, CAGGs, RLS policies, retention policies). Hand-author TimescaleDB extension DDL (hypertables, CAGGs, compression, retention) as raw SQL — drizzle-kit handles OLTP tables only. +- [ ] Write the initial Drizzle Kit migration for the full [target DDL](components/data-model.md#3-phase-0--schema-ddl-target-authoritative) (enums, entities, hypertables, CAGGs, dashboard rollup tables, RLS policies, retention policies). Hand-author TimescaleDB extension DDL (hypertables, CAGGs, compression, retention) as raw SQL — drizzle-kit handles OLTP tables only. +- [ ] **Instance-scoped uniqueness (F1):** every business-key unique is `UNIQUE (instance_id, …)` — `nodes.public_key`, `messages/advertisements/trace_paths.event_hash`, `channels.name/key_hex`; `settings` PK is `(instance_id, key)` +- [ ] **Only the two `raw_receptions`-sourced CAGGs (F2):** `cagg_daily_packet_counts`, `cagg_packet_breakdown_by_type`. Message/advert/node-count counts are worker-maintained rollup tables (`dashboard_daily_message_counts`, `dashboard_daily_advert_counts`, `dashboard_node_count_history`), NOT CAGGs +- [ ] **Hypertable node references are loose (no FK) (F6):** `raw_receptions.observer_node_id`, `event_observers.observer_node_id`, `telemetry.node_id`, `event_logs.observer_node_id` are plain `uuid` (avoids cross-chunk DML on node cleanup) +- [ ] **RLS hardening (F3):** every policy'd table has `FORCE ROW LEVEL SECURITY`; create the non-owner `meshcore_app` role (DML-only); the app connects as it, migrations run as the owner - [ ] Verify `drizzle-kit migrate` creates the schema cleanly on a fresh Postgres+TimescaleDB -- [ ] Verify RLS: a cross-instance query returns 0 rows +- [ ] Verify RLS **as the `meshcore_app` role**: a cross-instance query returns 0 rows (running as the owner proves nothing) - [ ] Seed the `instances` table from `NETWORK_NAME` +### NATS +- [ ] Provision the **single** platform-wide `INGEST` JetStream stream (subject `meshcore.ingest.>`) + one durable consumer `workers` — NOT a per-instance stream (F8) + ### Typed decoder models - [ ] Define Zod `DecodedPacket` schemas matching `@michaelhart/meshcore-decoder` output - [ ] Build the declarative `CLASSIFIERS` table (payload-type → event-type → handler) — [ingest.md §2](components/ingest.md#2-normalize-to-typed-envelopes-not-dicts) @@ -36,7 +47,7 @@ ### NATS infrastructure - [ ] Provision NATS with JetStream (file-backed persistence volume) -- [ ] Create the ingest stream (`INGEST-`, `duplicate_window=5m`, `max_age=7d`, `WorkQueuePolicy`) +- [ ] Configure the single `INGEST` stream (subject `meshcore.ingest.>`, `duplicate_window=5m`, `max_age=7d`, `WorkQueuePolicy`) — created in Phase 0 (F8) - [ ] Create the core fan-out subject pattern (`events.new..*`) - [ ] Create the channel-keys subject (`channel.keys..updated`) @@ -47,8 +58,9 @@ - [ ] Set `Nats-Msg-Id` = `wire_hash` for server-side dedup ### IngestWorker (batched write) -- [ ] Implement `IngestWorker.run`: pull-subscribe `meshcore.ingest.*`, fetch batches of 100, `SET LOCAL app.instance_id`, process, commit, publish `events.new`, ack -- [ ] Implement [`persist_deduped_event`](components/ingest.md#74-dedup-as-a-first-class-service) helper (SHA-256 hash, `ON CONFLICT DO NOTHING`, observer attach) +- [ ] Implement `IngestWorker.run`: pull-subscribe `meshcore.ingest.>` (wildcard, not `*` — F8), fetch batches of 100, `SET LOCAL app.instance_id`, process, commit, publish `events.new`, ack +- [ ] Implement [`persist_deduped_event`](components/ingest.md#74-dedup-as-a-first-class-service) helper (SHA-256 hash, `ON CONFLICT (instance_id, event_hash) DO NOTHING` — composite target, F1; observer attach) +- [ ] Implement `touchNode` + the observer-node upsert (both keyed on `(instance_id, public_key)`, F1/F11) - [ ] Implement the 4 structured handlers (~15 LOC each, using the dedup helper) - [ ] Implement the fallback `handle_event_log` handler @@ -57,20 +69,21 @@ - [ ] Wire the JSONPath-like filter DSL into production (evaluate `filter_expression` against event payload) - [ ] Verify webhook config reload on `settings.updated..webhooks` NATS notification -### Parallel-stack validation -- [ ] Stand up new stack alongside old, both subscribed to the same MQTT -- [ ] Build the diff harness (per-hour event counts by hash, old API vs new API) -- [ ] Validate for 5 days (D14); diff = 0 for 3 consecutive days to proceed +### Decode/classify shadow validation (F5 — DB-free) +- [ ] Run the `MqttIngester` against the live feed; diff its envelopes (decoded + classified output) against the old normalizer (24h shadow, no DB/workers) +- [ ] Full parallel-stack validation (both DBs, API diff) is **Phase 2** — it needs the provisioned schema + D5 outcome --- ## Phase 2 — Greenfield provisioning -### D5 benchmark (before schema freeze) -- [ ] Write `bench/route_match_benchmark.ts` — [testing.md → D5 plan](testing.md#d5-benchmark-plan-fold-vs-separate) -- [ ] Run at Low/Medium/High dataset shapes -- [ ] Record decision: fold (F) if `sweep_ms ≤ 1.5×` + `p95 ≤ 500ms` at High; else separate (S) -- [ ] Freeze schema accordingly +### D5 benchmark — done in Phase 0 (F5) +- [ ] (Moved to Phase 0 — the DDL already reflects the fold-vs-separate outcome.) Here: validate the route matcher against real ingested data at the D5 gate. + +### Full parallel-stack validation +- [ ] Stand up the new stack alongside the old, both subscribed to the same MQTT, both writing to their own DBs +- [ ] Build the diff harness (per-hour event counts; `wire_hash` coverage — NOT `event_hash`, since MD5≠SHA-256 across stacks — F4) +- [ ] Validate for 5 days (D14); diff = 0 for 3 consecutive days to proceed ### Config migration - [ ] Implement `meshcore-hub db export-config` on the old stack → JSON bundle @@ -87,11 +100,11 @@ - [ ] Implement `BlobStore` interface (`NoopBlobStore` default) — [ingest.md §5](components/ingest.md#5-raw-capture-compress-in-db-defer-object-storage) - [ ] D8 measurement: after 1 week of live data, check `hypertable_compression_stats('raw_receptions')` — activate object storage only if compressed size > 50% of DB and growth exceeds budget -### Continuous aggregates -- [ ] Create the 5 CAGGs (`WITH NO DATA`) via migration -- [ ] Add refresh policies (5-min schedule, 7-day window) -- [ ] Rewrite dashboard handlers to read CAGGs (no live-query fallback in greenfield) -- [ ] Verify first buckets populate within 10 min of live ingest +### Continuous aggregates + dashboard rollups (F2) +- [ ] Create the **2** CAGGs over `raw_receptions` (`cagg_daily_packet_counts`, `cagg_packet_breakdown_by_type`) `WITH NO DATA`; add refresh policies (5-min schedule, 7-day window) +- [ ] Create the **3** worker-maintained rollup tables (`dashboard_daily_message_counts`, `dashboard_daily_advert_counts`, `dashboard_node_count_history`) — sources are OLTP/entity tables, so they cannot be CAGGs +- [ ] Rewrite dashboard handlers to read CAGGs (with explicit `instance_id` predicate — RLS doesn't propagate to CAGGs) + the rollup tables (no live-query fallback in greenfield) +- [ ] Verify first CAGG buckets + rollup rows populate within 10 min of live ingest --- @@ -99,14 +112,14 @@ ### DerivedStateWorker - [ ] Implement `PeriodicJob` dataclass + `DerivedStateWorker` single-loop scheduler — [derived-state.md → Scheduler implementation](components/derived-state.md#scheduler-implementation) -- [ ] Register the 6 jobs: route-evaluator, route-history, spam-rescore, retention, metrics-gauges, cagg-health -- [ ] Implement `pg_advisory_xact_lock` per job (two-replica HA — D16) -- [ ] Verify two replicas don't double-execute the same job +- [ ] Register the 7 jobs: route-evaluator, route-history, spam-rescore, **dashboard-rollups** (F2), retention, metrics-gauges, cagg-health +- [ ] Implement the two-arg `pg_advisory_xact_lock(job_key, hashtext(instance_id))` per job — stable per-(job, instance) key, not a positional index (two-replica HA — D16; F7) +- [ ] Verify two replicas don't double-execute the same (job, instance) ### Spam scoring - [ ] Write the `compute_spam_score` PL/pgSQL function — [derived-state.md → Spam rescoring](components/derived-state.md#spam-rescoring-as-a-sql-function-online--sweep) - [ ] Wire it into the IngestWorker insert path (online score) -- [ ] Wire it into the `spam-rescore` job (symmetric sweep) +- [ ] Wire it into the `spam-rescore` job (symmetric sweep) — compute the score **once** per row in a subquery, filter + write from that value (do not call the function in both `WHERE` and `SET` — F10) - [ ] Verify parity: 24h replay, per-message score diff within ε ### Route health @@ -128,8 +141,9 @@ ### Async ORM - [ ] All route handlers are `async` with Drizzle ORM over `node-postgres` -- [ ] Connection pool sized for async concurrency; `SET LOCAL app.instance_id` hook on the pool -- [ ] Verify the pool correctly scopes transactions (advisory lock + RLS) +- [ ] **Per-request transaction (F3):** a Fastify `preHandler` opens a transaction and issues `SET LOCAL app.instance_id` from the `Principal` for **every** request (reads included — `SET LOCAL` outside a tx is a no-op → RLS returns 0 rows) +- [ ] App connects as the non-owner `meshcore_app` role so `FORCE ROW LEVEL SECURITY` applies +- [ ] Verify the pool correctly scopes transactions (advisory lock + RLS), including read endpoints ### Auth - [ ] Implement `AuthMiddleware` preHandler (JWT → cookie → API key → anonymous) — [auth.md](components/auth.md#authmiddleware-single-resolution-point) @@ -144,7 +158,7 @@ - [ ] Remove all `X-User-*` header injection ### Cache contract -- [ ] Implement the single `{namespace}:{scope}:{query_hash}` key format — [api.md → Unified cache contract](components/api.md#unified-cache-contract-concrete) +- [ ] Implement the single `{instance_id}:{namespace}:{scope}:{query_hash}` key format — the `instance_id` prefix is required so tenants never share a cache entry (F3) — [api.md → Unified cache contract](components/api.md#unified-cache-contract-concrete) - [ ] Implement the `NAMESPACES` / `ENTITY_INVALIDATION` declarative graph - [ ] Implement the async `@cached` decorator (ETag, If-None-Match, 304, X-Cache header) - [ ] Implement `invalidate_for(entity_changes, cache, instance_id)` diff --git a/docs/plans/next-gen/open-questions.md b/docs/plans/next-gen/open-questions.md index 8c9a897..8cf6679 100644 --- a/docs/plans/next-gen/open-questions.md +++ b/docs/plans/next-gen/open-questions.md @@ -52,6 +52,25 @@ All six iteration-5 review questions resolved: | Q6b | Token signing algorithm | **HS256 default**, RS256 as config option | | Q6c | Feature-flag propagation | **Next-boundary** per flag (next packet / next tick / next request), as documented in §8.8.3 | +## Resolved in iteration 8 (full-plan design review) + +A second full-plan review surfaced 13 issues — 4 schema-level blockers, 5 design risks, 4 softer notes — +all now corrected in place and catalogued in [review-findings.md](review-findings.md). The blockers were +resolved **before** the Phase 0 DDL freeze: + +| # | Issue | Resolution | +|---|---|---| +| F1 | Global unique constraints broke multi-tenancy (`nodes.public_key`, `event_hash`, `channels`, `settings` PK) | Instance-scoped composite uniques from Phase 0 — makes D21's "schema does not change" true | +| F2 | Two of five CAGGs can't be built (source tables aren't hypertables) | Only `raw_receptions`-sourced CAGGs remain; message/advert/node-count become worker-maintained rollup tables | +| F3 | RLS silently bypassable (owner bypass, autocommit reads, unscoped CAGGs + cache) | `FORCE RLS` + non-owner role, per-request transaction, `instance_id` in cache key, explicit CAGG predicate | +| F5 | D5 benchmark scheduled after the schema it decides | Moved to Phase 0; Phase 1 = decode shadow, Phase 2 = full parallel-stack | +| F4/F6/F7/F8 | Diff harness hash mismatch; compressed-hypertable FKs; unstable advisory-lock key; per-instance stream vs wildcard consumer | wire_hash join key; loose (FK-less) hypertable node refs; two-arg `pg_advisory_xact_lock(job, hashtext(instance_id))`; single `INGEST` stream | +| F10/F11 | Spam sweep double-eval; telemetry dedup race; missing observer upsert; rowid lookup | Single-eval sweep; documented best-effort telemetry dedup; observer node upsert; `raw_reception_received_at` for chunk exclusion | +| F9/F12/F13 | Python-shaped pseudocode; wizard SSR; scope/risk realism | TS-translation notes; wizard as SPA route; explicit strategy/risk note | + ## No remaining open design questions -The design covers Phases 0–7 concretely. All 22 decisions are locked. The remaining work is implementation, guided by the [phasing plan](phasing.md), [implementation checklist](implementation-checklist.md), and [testing/exit criteria](testing.md). +The design covers Phases 0–7 concretely. All 22 decisions are locked; iteration 8 corrected implementation +details without reopening any decision. The remaining work is implementation, guided by the +[phasing plan](phasing.md), [implementation checklist](implementation-checklist.md), +[testing/exit criteria](testing.md), and [review-findings.md](review-findings.md). diff --git a/docs/plans/next-gen/phasing.md b/docs/plans/next-gen/phasing.md index eef5499..bee1cdf 100644 --- a/docs/plans/next-gen/phasing.md +++ b/docs/plans/next-gen/phasing.md @@ -6,8 +6,9 @@ ## Phase 0 — Foundations (no behavior change) -- Fresh schema design; native `uuid`, enums, `JSONB`, SHA-256 hashes. -- Decide datastore strategy (D1, D2, D3, D4). +- **Run the D5 fold-vs-separate benchmark first** (synthetic data, throwaway TimescaleDB — testing.md → [D5 plan](testing.md#d5-benchmark-plan-fold-vs-separate)). Its outcome decides the `raw_receptions.path_hashes` shape, so it must land **before** the DDL is authored. This was previously (mis)scheduled in Phase 2, which created a circular dependency — the schema was "frozen" in Phase 0 but the benchmark that shapes it ran two phases later (F5). +- Fresh schema design (reflecting the D5 outcome); native `uuid`, enums, `JSONB`, SHA-256 hashes, instance-scoped composite uniques (F1), `FORCE ROW LEVEL SECURITY` + non-owner app role (F3). +- Decide datastore strategy (D1, D2, D3, D4). Provision the **single** platform-wide `INGEST` JetStream stream (`meshcore.ingest.>`) — not a per-instance stream (F8). - Stand up the typed `DecodedPacket` models + declarative classification table. - Set up the TS client tooling (orval config, `make gen-client`, CI drift check); first real generation happens in Phase 4 against the new API spec. @@ -19,7 +20,7 @@ - Split `MqttIngester` (pure decode+produce) from `IngestWorker` (batched write). - Centralize dedup helper; collapse handler boilerplate. - **`WebhookWorker`** (D19): NATS core subscriber on `events.new..*`; Tier-2 webhook settings; filter DSL wired into production. -- **Parallel-stack validation:** new stack ingests live MQTT alongside the old; diff outputs. +- **Decode/classify shadow validation:** run the `MqttIngester` against the live feed and diff its **envelopes** (decoded + classified output) against the old normalizer — no DB, no workers required. This isolates "does the new decode/classify match?" from the full end-to-end diff. The full **parallel-stack** validation (both stacks writing to their DBs, compared at the API) is a Phase 2 activity, because it needs the provisioned schema and the D5 outcome — running it here would make Phase 1 depend on Phase 2 (F5). > **Detailed design:** [components/ingest.md](components/ingest.md), [components/infrastructure.md](components/infrastructure.md). Exit criteria: [Phase 1](testing.md#phase-1--ingest-pipeline). @@ -27,7 +28,8 @@ - **Greenfield infra:** fresh Postgres+TimescaleDB, NATS, new schema. No historical data migration. - **Preserved-config export/import** (`db export-config` / `db import-config`): user_profiles + roles, routes + nodes + observers, node_tags, adoptions, channels, plus node identity stubs. -- **D5 spike:** fold `packet_path_hops` into `raw_receptions.path_hashes` array; benchmark route matcher; keep folded if perf holds, else keep a hypertable. +- **Full parallel-stack validation** (moved here from Phase 1): both stacks ingest the same live MQTT into their own DBs; the diff harness compares per-hour event counts and `wire_hash` coverage (F4) at the API. This is the DB-level gate; the decode-level shadow was Phase 1. +- **D5 outcome applied** (the benchmark itself ran in Phase 0, F5): the schema already reflects fold-vs-separate; here the route matcher is validated against real data at the D5 gate. - **D8 step 1:** keep `raw_hex` in-DB but rely on TimescaleDB compression (10–20×). **D8 step 2 (only if measured necessary):** move bytes to a `BlobStore` (MinIO/local-volume) behind an interface. - No historical data migration — preserved config only (see [migration.md](components/migration.md)). @@ -36,7 +38,7 @@ ## Phase 3 — Derived state consolidation - Replace the 6 background threads with the single `DerivedStateWorker`. -- Convert **dashboard aggregations** to TimescaleDB continuous aggregates (the pure time-bucketing ones). +- Convert the **hypertable-sourced** dashboard aggregations (daily packet counts, packet breakdown by type — over `raw_receptions`) to TimescaleDB continuous aggregates. The message/advert/node-count aggregations source from OLTP/entity tables and **cannot be CAGGs** — they become worker-maintained rollup tables via the `dashboard-rollups` job (F2). - Route health: rewrite the matcher against `raw_receptions.path_hashes` (D5 outcome); collapse the 3 derived tables into worker-maintained state. - Move spam rescoring to a DB function + periodic sweep. @@ -124,9 +126,35 @@ After the single-tenant stack (Phases 0–6) is stable, extend to shared-platfor | Local-password auth becomes a brute-force target | argon2id + exponential lockout (§8.3.2) + reverse-proxy rate limiting | | D5 fold benchmark fails (separate table needed) | Reversible — `raw_receptions.path_hashes` stays either way; hops table is additive | | Scope creep | Each phase is independently valuable and shippable; we can stop after any phase | +| **Highest-risk migration shape:** greenfield rewrite + language switch (Python→TS) + two new infra deps (NATS, TimescaleDB) + multi-tenancy, all in one program | Sequenced so the language/infra risk is front-loaded (Phases 0–1) and de-risked by the decode-shadow (Phase 1) + parallel-stack (Phase 2) gates before any cutover; multi-tenancy is deferred to Phase 7 (fully additive). See the strategy note below. | --- +## Strategy note — scope realism (F13) + +This is a candid caveat, not a decision reversal. The plan is a **from-scratch greenfield rewrite** of a +mature, feature-complete product that *also* switches backend language (Python→TypeScript, D22), adopts +two new infrastructure dependencies (NATS, TimescaleDB), and adds multi-tenancy. That is the highest-risk +combination on the migration-strategy spectrum, and two framing claims deserve qualification: + +- **"Each phase is independently valuable and shippable" is true for *value*, not for *production*.** + Phases 0–5 are not individually production-serviceable: the ingest pipeline (Phase 1) has no schema to + write into until Phase 2's provisioning; the API (Phase 4) is not useful without auth; the frontend + (Phase 5) needs the API. The honest statement is that the *program* is shippable at the Phase 6 cutover, + with earlier phases being internally verifiable milestones. "We can stop after any phase" means we can + stop *building*, not that an intermediate phase is a deployable product. +- **The strangler-fig alternative was considered and rejected in favour of greenfield** (the iteration-4 + greenfield decision). Worth keeping visible: a strangler approach — the new TS pipeline writing into the + *existing* database, swapped in component-by-component behind the running Python app — trades the + greenfield's clean schema for a lower-risk cutover (no big-bang, no 5-day all-or-nothing window). The + greenfield was chosen because it eliminates the backfill subsystem entirely; the cost is that + correctness across the whole feature surface must be proven within the parallel-stack window rather than + incrementally in production. + +No timeline or effort estimate is asserted here; the surface (19 tables, ~13 API routers, the full SPA, a +new language, new infra, multi-tenancy) is large, and the phase list is a dependency order, not a +schedule. + ## Design retrospective (what shifted across iterations 1–7) | Iteration | What was proposed | What changed | Why | diff --git a/docs/plans/next-gen/review-findings.md b/docs/plans/next-gen/review-findings.md new file mode 100644 index 0000000..2e556e1 --- /dev/null +++ b/docs/plans/next-gen/review-findings.md @@ -0,0 +1,146 @@ +# Design Review Findings & Resolutions (iteration 8) + +> A full-plan review after iteration 7 surfaced 13 issues. This document records each finding, +> its resolution, and the files changed. The corrections are applied in-place across the plan; +> the four schema-level items (F1–F3, F5) were treated as blockers to the Phase 0 DDL freeze. + +## Blockers (schema / correctness — resolved before DDL freeze) + +### F1 — Multi-tenant uniqueness was single-tenant-only ("schema does not change" was false) +Several uniqueness constraints in the Phase 0 DDL were **global**, contradicting D21/multi-tenancy.md's +claim that the schema is already instance-scoped: + +- `nodes.public_key UNIQUE` — physical nodes are shared across communities by RF; two tenants + could not both hold a row, and `touchNode`'s `ON CONFLICT (public_key)` would collide. +- `messages/advertisements/trace_paths.event_hash UNIQUE` — global, so tenant B silently never + ingests an event tenant A already dedup'd (multi-tenancy.md §10 promises one event *per tenant*). +- `channels.name` / `channels.key_hex UNIQUE` — two communities could not share a channel name. +- `settings` PK was `key` alone — exactly one settings row per key **across the whole platform**. + +**Resolution:** all of these are now **instance-scoped composite keys from Phase 0** +(`UNIQUE (instance_id, …)`, `settings PRIMARY KEY (instance_id, key)`). In single-tenant mode there is +one instance, so behaviour is identical; in Phase 7 the schema is genuinely additive. The dedup helper +and `touchNode` now target the composite keys. D21 / multi-tenancy.md reworded: the schema is built +instance-scoped from Phase 0 — Phase 7 adds tables, not constraint migrations. +Files: `components/data-model.md`, `components/ingest.md`, `components/multi-tenancy.md`, +`components/api.md`, `decisions/D21-multi-tenancy.md`. + +### F2 — Two of the five continuous aggregates could not be created +`cagg_daily_message_counts` (over `messages`) and `cagg_daily_advert_counts` (over `advertisements`) +were declared as TimescaleDB continuous aggregates, but CAGGs require a **hypertable** source and both +tables are deliberately plain OLTP (uuid PK, content-hash dedup). Promoting them to hypertables would +break `event_hash` dedup (a hypertable unique must include the partition column). `cagg_node_count_history` +has no append-only time-series source either. + +**Resolution:** only the two `raw_receptions`-sourced CAGGs remain (`cagg_daily_packet_counts`, +`cagg_packet_breakdown_by_type`). Daily message counts, advert counts, and node-count history become +**worker-maintained dashboard rollup tables** (instance-scoped, RLS'd), refreshed by a new +`dashboard-rollups` DerivedStateWorker job — the same "can't be a CAGG, so the worker owns it" pattern +already used for route health. Files: `components/data-model.md`, `components/derived-state.md`, +`components/api.md`, `phasing.md`, `testing.md`, `overview.md`. + +### F3 — RLS had three silent-bypass / leakage gaps +1. **Owner bypass** — RLS is skipped for the table owner unless `FORCE ROW LEVEL SECURITY` is set and + the app connects as a non-owner role. Neither was specified. +2. **`SET LOCAL` needs a transaction** — `current_setting('app.instance_id')` is NULL for any statement + run in autocommit, so every read (not just the worker's writes) must run inside a transaction that + sets the GUC. The read path never showed this. +3. **CAGGs and the response cache were not instance-scoped** — RLS does not propagate to continuous + aggregates, and the cache key format (`{namespace}:{scope}:{query_hash}`) omitted `instance_id`, + so tenants shared cache entries and `invalidate_for` flushed all tenants. + +**Resolution:** added `FORCE ROW LEVEL SECURITY` + a dedicated non-owner `meshcore_app` role to the RLS +template; documented a per-request transaction that issues `SET LOCAL app.instance_id` for reads and +writes; dashboard CAGG reads now carry an explicit `instance_id` predicate; cache key is now +`{instance_id}:{namespace}:{scope}:{query_hash}` and `invalidate_for` deletes are instance-scoped. +Files: `components/data-model.md`, `decisions/D03-row-level-tenancy-rls.md`, `components/api.md`, +`components/multi-tenancy.md`. + +### F5 — Circular phase dependency around the D5 schema decision +The D5 fold-vs-separate benchmark (which decides the `raw_receptions.path_hashes` shape) was scheduled in +Phase 2, but Phase 0 writes the full DDL migration and Phase 1's parallel-stack validation writes into +that schema — so the schema was "frozen" two phases before the benchmark that shapes it, and Phase 1 +depended on Phase 2 outputs. + +**Resolution:** the D5 benchmark moves to **Phase 0** (synthetic-data-only, needs only a throwaway +TimescaleDB — no cross-phase dependency), and the schema migration is authored after the outcome, still +within Phase 0. Phase 1 is now the **decode/classify shadow** (MqttIngester envelope diff, no DB); the +full **parallel-stack** validation (DB + workers + API diff) is Phase 2. Files: `phasing.md`, +`decisions/D05-fold-packet-path-hops.md`, `implementation-checklist.md`, `components/infrastructure.md`, +`testing.md`, `components/migration.md`. + +## Design risks (resolved with corrections) + +### F6 — FKs from compressed hypertables to `nodes` + hourly node cleanup +`raw_receptions.observer_node_id`, `event_observers.observer_node_id`, `telemetry.node_id`, +`event_logs.observer_node_id` were FKs to `nodes` with `ON DELETE SET NULL/CASCADE`. Hourly +`cleanup_inactive_nodes` deletes node rows, forcing SET NULL/CASCADE DML across hypertable chunks — +including **compressed** ones, where DML is restricted/costly. + +**Resolution:** those columns are now **loose `uuid` references (no FK)**, matching the existing +`route_recent_matches.raw_reception_rowid` precedent and the plan's already-accepted tolerance for +orphaned hypertable rows (they compress and age out on retention). Files: `components/data-model.md`, +`components/derived-state.md`. + +### F7 — Unstable per-instance advisory-lock key (double execution under HA) +`lock_key = base_key + instance_index` used a positional index that shifts when tenants are added/removed +and can differ between replicas, defeating D16's single-execution guarantee. + +**Resolution:** the DerivedStateWorker now uses the two-argument `pg_advisory_xact_lock(job_key, +hashtext(instance_id))` — a stable per-(job, instance) key. Files: `components/multi-tenancy.md`, +`components/derived-state.md`, `decisions/D16-two-replica-worker-ha.md`. + +### F8 — NATS per-instance stream vs. cross-tenant wildcard consumer +D4/ingest.md defined a per-instance `INGEST-` WorkQueuePolicy stream, but Phase 7's shared worker +pool subscribes to `meshcore.ingest.>` (all tenants). A consumer group cannot span multiple streams, and +the single-tenant worker example subscribed to `meshcore.ingest.*` (a token-count mismatch for the +4-token subject). + +**Resolution:** one shared `INGEST` stream captures `meshcore.ingest.>`; a single durable consumer +`workers` is shared by all IngestWorker replicas; single-tenant is just one instance's subjects. +Worker subscribe corrected to `meshcore.ingest.>`. Files: `components/ingest.md`, +`decisions/D04-nats-jetstream-ingest.md`, `components/infrastructure.md`, `components/multi-tenancy.md`. + +### F4 — Diff harness could not match events by hash +The harness compared `event_hash` between stacks, but the old stack uses MD5 and the new uses SHA-256, +so the same event has different hashes and coverage is always 0%. + +**Resolution:** the hash-coverage check keys on `wire_hash` (the LetsMesh on-air hash, identical in both +stacks). Files: `components/migration.md`, `testing.md`. + +### F10 — Spam-rescore sweep evaluated the scoring function twice per row +The sweep `WHERE spam_score IS DISTINCT FROM compute_spam_score(...)` plus `SET spam_score = +compute_spam_score(...)` ran the (COUNT-heavy) PL/pgSQL function twice per candidate every 120s. + +**Resolution:** the sweep computes the score once in a subquery/CTE and filters + writes from that single +value. Files: `components/derived-state.md`. + +### F11 — Smaller gaps +- **Telemetry dedup race** across concurrent worker replicas (no unique constraint possible on the + hypertable) — documented as best-effort, backed by `Nats-Msg-Id` window dedup + read-side de-dup by + `event_hash`. (`components/data-model.md`, `components/ingest.md`) +- **Observer (receiver) node upsert** was missing — `touchNode` now also find-or-creates the observing + node before the FK-less reception/junction rows reference it. (`components/ingest.md`) +- **`route_recent_matches`** now stores `raw_reception_received_at` so match lookups get chunk exclusion + instead of scanning every chunk. (`components/data-model.md`) +- **`telemetry` unbounded growth** — added an optional retention policy tied to a tuning setting. + (`components/data-model.md`, `components/derived-state.md`) + +## Notes / softer points (documented, not redesigned) + +### F9 — Component-doc code is Python-shaped; some patterns don't survive the D22 TS switch +The illustrative snippets use `selectinload`, FastAPI `Depends` injection, and an `@cached` decorator, +which have no 1:1 Drizzle/Fastify equivalent. Added an explicit **TS translation note** in `api.md` +mapping them to Drizzle relational queries, Fastify `preHandler` hooks, and a cache plugin/hook, so the +sections aren't under-specified for implementation. + +### F12 — First-run setup wizard reintroduces server-rendered HTML +The SSR wizard contradicts the static-shell principle (Principle 5). Documented the preferred approach — +a normal SPA route gated by a `needsSetup` flag in `/api/v1/config` — with SSR kept only as a fallback. +Files: `components/auth.md`. + +### F13 — Greenfield rewrite + language switch + new infra + multi-tenancy is the highest-risk path +Added an explicit risk row and a short strategy note: Phases 0–5 are not independently *production*- +shippable (only collectively), the strangler-fig alternative was considered, and multi-tenancy (Phase 7) +and the language switch sit on the critical path. No timeline/effort estimate is asserted. Files: +`phasing.md`. diff --git a/docs/plans/next-gen/testing.md b/docs/plans/next-gen/testing.md index 8ad60cd..12e145e 100644 --- a/docs/plans/next-gen/testing.md +++ b/docs/plans/next-gen/testing.md @@ -13,12 +13,16 @@ - [ ] Server-side dedup (`Nats-Msg-Id`) suppresses MQTT redelivery with zero double-inserts. - [ ] `WebhookWorker` dispatches matching events with correct retry/backoff; filter DSL evaluates correctly; config reload on `settings.updated.webhooks` works. +## Phase 0 — Foundations (added exit gate) + +- [ ] **D5 benchmark decided fold-vs-separate**, decision recorded, target schema frozen accordingly — this runs in Phase 0, **before** the DDL migration is authored (it needs only a throwaway TimescaleDB; see [D5 plan](#d5-benchmark-plan-fold-vs-separate)). The schema is not frozen until the outcome is known. + ## Phase 2 — Greenfield provisioning -- [ ] D5 benchmark decided fold-vs-separate, decision recorded, target schema frozen accordingly. - [ ] `db export-config` on the old stack produces a complete bundle; `db import-config` on a fresh DB reproduces user_profiles + roles, routes + nodes + observers, node_tags, adoptions, (channels) with zero FK violations. -- [ ] New infrastructure provisioned; `drizzle-kit migrate` creates the full schema cleanly; RLS policies enforced (a cross-instance query returns 0 rows). -- [ ] All five CAGGs created with active refresh policies; first buckets populate within 10 minutes of live ingest. +- [ ] New infrastructure provisioned; `drizzle-kit migrate` creates the full schema cleanly; RLS policies enforced **as the non-owner `meshcore_app` role with `FORCE ROW LEVEL SECURITY`** (a cross-instance query returns 0 rows; verify the check runs as the app role, not the table owner). +- [ ] The **two** CAGGs (`cagg_daily_packet_counts`, `cagg_packet_breakdown_by_type`, over `raw_receptions`) created with active refresh policies; the **three** worker-maintained rollup tables (`dashboard_daily_message_counts`, `dashboard_daily_advert_counts`, `dashboard_node_count_history`) populated by the `dashboard-rollups` job; first buckets/rows populate within 10 minutes of live ingest. +- [ ] Dashboard CAGG reads carry an explicit `instance_id` predicate (RLS does not propagate to continuous aggregates); rollup tables enforce RLS like any tenant table. - [ ] Hypertable compression + retention policies active and verified on all four hypertables (`raw_receptions`, `event_observers`, `telemetry`, `event_logs`): drop a chunk manually, confirm rows go; verify `event_observers` segments by `event_type` (query by event_type hits compressed batches correctly). - [ ] Parallel-stack diff harness reports 0 divergence for 3 consecutive days within the 5-day window (D14). @@ -30,7 +34,7 @@ - [ ] Route health tables rebuild correctly from `raw_receptions.path_hashes`; per-route `total_route_ms` within the D5 gate. - [ ] `quality_avg` matches today's output on a 7-day replay: for each route, the rolling 7-day ordinal average (clear=2, marginal=1, else=0; thresholds ≥1.5/≥0.75) produces the same tier as the old `compute_persisted_quality_avg`. - [ ] Retention enforces 30-day windows via chunk drops (verify `raw_receptions` row count stabilises). -- [ ] Dashboard endpoints read CAGGs exclusively; live-query fallback removed. +- [ ] Dashboard endpoints read the CAGGs + the worker-maintained rollup tables exclusively; live-query fallback removed. The `dashboard-rollups` job maintains the message/advert/node-count rollups. ## Phase 4 — API & auth @@ -121,4 +125,4 @@ If folded fails the gate: fall back to the separate hypertable (still better tha #### Timing -Budget half a day. Record results in `docs/plans/-d5-fold-benchmark/` with the dataset parameters and the decision. **Run this before the new stack's schema is frozen**, so the DDL reflects the outcome. +Budget half a day. Record results in `docs/plans/-d5-fold-benchmark/` with the dataset parameters and the decision. **Run this in Phase 0, before the new stack's schema is frozen**, so the DDL migration reflects the outcome (it needs only a throwaway TimescaleDB — no dependency on any later phase).