Coverage for node / src / stigmem_node / subscription_delivery.py: 89%
188 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Subscription delivery engine — spec §20.4.
3Responsibilities:
41. ``fan_out`` — when a fact is written, find matching subscriptions and insert
5 pending ``subscription_events`` rows (fast; called synchronously on every write).
62. ``deliver_pending`` — sweep: attempt delivery for all pending/due-for-retry events.
73. ``sweep_loop`` — asyncio task run in the app lifespan; calls deliver_pending periodically.
9Delivery types:
10- ``webhook`` — HTTP POST to delivery_address with exponential backoff (1 s, 2 s, 4 s … ≤ 300 s).
11- ``wake`` — structured JSON emitted to stderr; platform-specific integration hook.
13Garden ACL (§17) and content sanitizer (§19) are applied at delivery time.
14"""
16from __future__ import annotations
18import asyncio
19import json
20import logging
21import sys
22import threading
23import time
24import uuid
25from datetime import UTC, datetime
26from typing import Any
27from urllib.parse import urlsplit, urlunsplit
29import httpx
31from . import settings as _settings_pkg
32from .auth import Identity
33from .db import db
34from .garden_acl import get_member_role
35from .lifecycle.tombstone_cache import is_tombstoned as _is_tombstoned
36from .memory_garden_acl_gate import garden_acl_enforced
37from .net_util import resolve_pinned_address
38from .recall.recall_pipeline import apply_recall_pipeline
40logger = logging.getLogger("stigmem.subscriptions")
41_DELIVER_PENDING_LOCK = threading.Lock()
44# ---------------------------------------------------------------------------
45# Fan-out — called from routes/facts.py after each successful fact write
46# ---------------------------------------------------------------------------
49def fan_out(
50 fact_id: str,
51 entity: str,
52 scope: str,
53 garden_id: str | None,
54 tenant_id: str,
55 fact_payload_json: str,
56) -> None:
57 """Find subscriptions matching this fact and create pending delivery events."""
58 now = datetime.now(UTC).isoformat()
60 with db() as conn:
61 scope_subs = conn.execute(
62 """SELECT * FROM subscriptions
63 WHERE target_kind='scope' AND target=? AND tenant_id=? AND circuit_open=0""",
64 (scope, tenant_id),
65 ).fetchall()
67 entity_subs = conn.execute(
68 """SELECT * FROM subscriptions
69 WHERE target_kind='entity' AND target=? AND tenant_id=? AND circuit_open=0""",
70 (entity, tenant_id),
71 ).fetchall()
73 for sub in list(scope_subs) + list(entity_subs):
74 event_id = str(uuid.uuid4())
75 conn.execute(
76 """INSERT INTO subscription_events
77 (id, subscription_id, event_type, entity_uri, fact_id,
78 payload, created_at, delivery_status)
79 VALUES (?,?,?,?,?,?,?,'pending')""",
80 (event_id, sub["id"], "fact_asserted", entity, fact_id, fact_payload_json, now),
81 )
84# ---------------------------------------------------------------------------
85# Delivery sweep — called by sweep_loop every N seconds
86# ---------------------------------------------------------------------------
89def deliver_pending() -> None:
90 """Attempt delivery for all pending events that are due.
92 Uses an atomic claim (``pending → delivering``) so that the background
93 ``sweep_loop`` and any concurrent caller (admin replay endpoints,
94 tests, future per-tenant workers) cannot deliver the same event twice.
95 See issue #47 and migration 028 for the rationale.
97 Concurrency model
98 -----------------
99 1. Recover stale claims: any row in ``delivering`` with
100 ``claimed_at`` older than ``subscription_claim_timeout_s`` is reset
101 to ``pending`` so a crashed worker cannot strand events.
102 2. Atomically claim up to 100 due rows via
103 ``UPDATE … WHERE id IN (SELECT … LIMIT 100) RETURNING …`` — SQLite
104 serializes writes, so concurrent claimers see disjoint row sets.
105 3. For each claimed row, attempt delivery and transition to
106 ``delivered`` (success) or back to ``pending`` with a fresh
107 ``next_retry_at`` (failure).
108 """
109 if not _DELIVER_PENDING_LOCK.acquire(blocking=False):
110 logger.debug("subscription delivery already in progress; skipping concurrent drain")
111 return
113 try:
114 now = datetime.now(UTC).isoformat()
115 claim_timeout_s = _settings_pkg.settings.subscription_claim_timeout_s
116 stale_cutoff = datetime.fromtimestamp(
117 time.time() - claim_timeout_s,
118 UTC,
119 ).isoformat()
121 with db() as conn:
122 # 0. Prune terminal events older than the retention window (M12 / F-AVAIL-2).
123 # Only delivered/failed rows are touched — pending/delivering are never pruned.
124 # The horizon is clamped to >= subscription_replay_s so the replay API
125 # never loses events it is supposed to surface.
126 retention_s = _settings_pkg.settings.subscription_event_retention_s
127 if retention_s > 0:
128 effective_retention_s = max(
129 retention_s,
130 _settings_pkg.settings.subscription_replay_s,
131 )
132 prune_cutoff = datetime.fromtimestamp(
133 time.time() - effective_retention_s,
134 UTC,
135 ).isoformat()
136 conn.execute(
137 """DELETE FROM subscription_events
138 WHERE delivery_status IN ('delivered', 'failed')
139 AND created_at < ?""",
140 (prune_cutoff,),
141 )
143 # 1. Recover stale claims left behind by a crashed worker. We do NOT
144 # reset delivery_attempts — the next attempt counts as a retry.
145 conn.execute(
146 """UPDATE subscription_events
147 SET delivery_status='pending', claimed_at=NULL
148 WHERE delivery_status='delivering'
149 AND claimed_at IS NOT NULL
150 AND claimed_at < ?""",
151 (stale_cutoff,),
152 )
154 # 2. Atomic claim. The inner SELECT is the same predicate the old
155 # non-atomic path used, joined to subscriptions for circuit-open
156 # filtering. The outer UPDATE … RETURNING gives us the claimed
157 # rows in a single round-trip; no other caller can claim them
158 # until we release them.
159 claimed = conn.execute(
160 """UPDATE subscription_events
161 SET delivery_status='delivering', claimed_at=?
162 WHERE id IN (
163 SELECT e.id
164 FROM subscription_events e
165 JOIN subscriptions s ON e.subscription_id = s.id
166 WHERE e.delivery_status = 'pending'
167 AND s.circuit_open = 0
168 AND (e.next_retry_at IS NULL OR e.next_retry_at <= ?)
169 ORDER BY e.created_at ASC
170 LIMIT 100
171 )
172 RETURNING id""",
173 (now, now),
174 ).fetchall()
176 if not claimed:
177 return
179 claimed_ids = [row["id"] for row in claimed]
180 # ``placeholders`` is a fixed "?,?,?…" string whose length comes from
181 # ``claimed_ids`` (UUIDs we just emitted into our own table) — no user
182 # input flows into the SQL text, so the f-string interpolation is safe.
183 placeholders = ",".join("?" * len(claimed_ids))
184 _base = (
185 "SELECT e.id, e.subscription_id, e.event_type, e.entity_uri, e.fact_id,"
186 " e.payload, e.created_at, e.delivery_attempts,"
187 " s.on_change, s.delivery_address, s.subscriber_identity, s.tenant_id,"
188 " s.circuit_open"
189 " FROM subscription_events e"
190 " JOIN subscriptions s ON e.subscription_id = s.id"
191 " WHERE e.id IN ("
192 )
193 select_sql = _base + placeholders + ")" # noqa: S608 — fixed "?,…" string
194 events = conn.execute(select_sql, claimed_ids).fetchall()
196 for event in events:
197 try:
198 payload = json.loads(event["payload"])
199 success = _deliver_one(event, payload)
200 except Exception as exc:
201 logger.error("Delivery error for event %s: %s", event["id"], exc)
202 success = False
204 _record_result(event, success)
205 finally:
206 _DELIVER_PENDING_LOCK.release()
209# ---------------------------------------------------------------------------
210# Delivery helpers
211# ---------------------------------------------------------------------------
214def _deliver_one(event: Any, payload: dict[str, Any]) -> bool:
215 on_change = event["on_change"]
216 if on_change == "webhook":
217 return _deliver_webhook(event, payload)
218 if on_change == "wake": 218 ↛ 220line 218 didn't jump to line 220 because the condition on line 218 was always true
219 return _deliver_wake(event, payload)
220 logger.error("Unknown on_change %r for event %s", on_change, event["id"])
221 return False
224def _subscriber_identity(entity_uri: str, tenant_id: str) -> Identity:
225 return Identity(entity_uri=entity_uri, permissions=["read"], tenant_id=tenant_id)
228def _subscriber_has_active_key(entity_uri: str, tenant_id: str) -> bool:
229 """Return True if at least one non-expired API key exists for this identity.
231 Only consulted when STIGMEM_AUTH_REQUIRED=true so that single-operator
232 (auth-disabled) nodes are not affected.
233 """
234 from . import settings as _sp
236 if not _sp.settings.auth_required: 236 ↛ 238line 236 didn't jump to line 238 because the condition on line 236 was always true
237 return True
238 now = datetime.now(UTC).isoformat()
239 with db() as conn:
240 row = conn.execute(
241 "SELECT id FROM api_keys WHERE entity_uri=? AND tenant_id=?"
242 " AND (expires_at IS NULL OR expires_at > ?) LIMIT 1",
243 (entity_uri, tenant_id, now),
244 ).fetchone()
245 return row is not None
248def _sanitize_payload(event: Any, payload: dict[str, Any]) -> dict[str, Any] | None:
249 """Apply §17 garden ACL and §19 sanitizer. Returns None to suppress delivery."""
250 from .models.facts import FactRecord, FactValue
252 subscriber = event["subscriber_identity"]
253 tenant_id = event["tenant_id"]
255 # S2 fix: re-check that the subscriber still holds an active read credential.
256 # Prevents continued delivery after API key revocation for scope/entity subscriptions.
257 if not _subscriber_has_active_key(subscriber, tenant_id): 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 return None
260 # §17 garden ACL re-check: skip delivery if subscriber no longer a member
261 garden_uuid = payload.get("garden_id")
262 if garden_uuid and garden_acl_enforced():
263 role = get_member_role(garden_uuid, subscriber)
264 if role is None: 264 ↛ 269line 264 didn't jump to line 269 because the condition on line 264 was always true
265 return None
267 # §23.3.3 r.2: drop event immediately when entity has an active tombstone.
268 # Uses the in-process cache (§23.3.3 r.4); 60-second leak window is acceptable.
269 entity_for_tombstone = payload.get("entity")
270 if entity_for_tombstone and _is_tombstoned(entity_for_tombstone, tenant_id):
271 return None
273 try:
274 record = FactRecord(
275 id=payload["id"],
276 entity=payload["entity"],
277 relation=payload["relation"],
278 value=FactValue(type=payload["value_type"], v=payload.get("value_v")),
279 source=payload["source"],
280 timestamp=payload["timestamp"],
281 confidence=float(payload.get("confidence", 1.0)),
282 scope=payload.get("scope", "local"),
283 )
284 identity = _subscriber_identity(subscriber, tenant_id)
285 results = apply_recall_pipeline([record], identity=identity, include_low_trust=True)
286 if not results: 286 ↛ 287line 286 didn't jump to line 287 because the condition on line 286 was never true
287 return None
288 sanitized = results[0]
289 if sanitized.sanitizer_redacted: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 return {"fact_id": payload["id"], "redacted": True}
291 out = dict(payload)
292 out["value_v"] = str(sanitized.value.v) if sanitized.value.v is not None else None
293 if sanitized.sanitizer_warnings:
294 out["sanitizer_warnings"] = sanitized.sanitizer_warnings
295 return out
296 except Exception as exc:
297 logger.warning("Sanitizer error for fact %s: %s", payload.get("id"), exc)
298 return payload
301def _build_pinned_request(address: str, pinned_ip: str) -> tuple[str, str]:
302 """Return ``(pinned_url, host_header)`` for connecting to *pinned_ip*.
304 The returned URL swaps the original hostname for the validated *pinned_ip*
305 literal (IPv6 bracketed) while preserving scheme/port/path/query, so the
306 socket connects to the pinned IP and cannot be re-resolved by a rebinder.
307 The ``host_header`` carries the ORIGINAL hostname (+ explicit port) so the
308 webhook receiver still sees the name it was registered under.
310 The caller also passes ``extensions={"sni_hostname": <hostname>}`` so the TLS
311 SNI and certificate verification run against the original hostname, NOT the
312 IP literal (proven against httpx 0.28.1 in test_webhook_dns_rebind.py).
313 """
314 parts = urlsplit(address)
315 hostname = parts.hostname or ""
316 port = parts.port
317 # Bracket IPv6 literals for a valid authority.
318 ip_authority = f"[{pinned_ip}]" if ":" in pinned_ip else pinned_ip
319 if port is not None:
320 netloc = f"{ip_authority}:{port}"
321 host_header = f"{hostname}:{port}"
322 else:
323 netloc = ip_authority
324 host_header = hostname
325 pinned_url = urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
326 return pinned_url, host_header
329def _deliver_webhook(event: Any, payload: dict[str, Any]) -> bool:
330 sanitized = _sanitize_payload(event, payload)
331 if sanitized is None:
332 # ACL/sanitizer blocked — mark delivered, don't retry
333 _mark_delivered(event["id"], event["subscription_id"])
334 return True
336 # SSRF guard (P-CONF-2 + H9/F-SSRF-1): never POST to a private/loopback/
337 # link-local/IMDS address. Resolve the hostname ONCE and PIN the connection to
338 # that validated IP — assert_safe_url's resolved-then-reconnect pattern left a
339 # DNS-rebind TOCTOU window (httpx re-resolved the hostname at connect time).
340 # resolve_pinned_address rejects the whole URL if ANY resolved record is
341 # private (the rebinder picks which record is served) and returns one safe IP.
342 # Unsafe/unresolvable addresses can never become deliverable, so stop retrying.
343 # GHSA-5p3m-vhh6-9236: https-only by default; http requires the explicit
344 # operator opt-in (STIGMEM_WEBHOOK_ALLOW_INSECURE_HTTP).
345 address = event["delivery_address"]
346 try:
347 pinned_ip = resolve_pinned_address(
348 address,
349 allow_schemes=_settings_pkg.settings.webhook_allowed_schemes,
350 )
351 except ValueError as exc:
352 logger.warning(
353 "Webhook delivery blocked for subscription %s: unsafe delivery_address (%s)",
354 event["subscription_id"],
355 exc,
356 )
357 _mark_delivered(event["id"], event["subscription_id"])
358 return True
360 # Connect to the pinned IP literal (no re-resolution) while verifying the TLS
361 # cert against the ORIGINAL hostname via the sni_hostname extension, and carry
362 # the original Host header. Proven against httpx 0.28.1 in
363 # tests/routes/test_webhook_dns_rebind.py (a real TLS handshake verifies the
364 # cert against the hostname, not the IP).
365 pinned_url, host_header = _build_pinned_request(address, pinned_ip)
366 hostname = urlsplit(address).hostname or ""
368 body = {
369 "event_id": event["id"],
370 "idempotency_key": event["id"],
371 "subscription_id": event["subscription_id"],
372 "event_type": event["event_type"],
373 "fact": sanitized,
374 }
376 try:
377 with httpx.Client(timeout=10.0, follow_redirects=False) as client:
378 resp = client.post(
379 pinned_url,
380 json=body,
381 headers={
382 "Content-Type": "application/json",
383 "X-Stigmem-Event-Id": event["id"],
384 "Host": host_header,
385 },
386 extensions={"sni_hostname": hostname},
387 )
389 if resp.status_code == 410:
390 # Webhook endpoint gone — cancel the subscription
391 with db() as conn:
392 conn.execute("DELETE FROM subscriptions WHERE id=?", (event["subscription_id"],))
393 logger.info("Subscription %s cancelled (410 Gone)", event["subscription_id"])
394 return True
396 # 5xx / 429 → retry; 2xx/3xx → success; 4xx (except 429) → permanent failure, don't retry
397 if resp.status_code >= 500 or resp.status_code == 429:
398 return False
399 return resp.status_code < 400
401 except (httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError):
402 return False
405def _deliver_wake(event: Any, payload: dict[str, Any]) -> bool:
406 sanitized = _sanitize_payload(event, payload)
407 if sanitized is None: 407 ↛ 408line 407 didn't jump to line 408 because the condition on line 407 was never true
408 return True # ACL blocked; mark delivered
410 # P1 note: wake delivery writes sanitized fact payloads to stderr for
411 # operator/platform pickup. Any process with stderr access sees ALL wake
412 # events from ALL subscribers. This is intentional for the operator
413 # integration use-case but means the platform operator is implicitly
414 # trusted with the content of every wake-delivered fact. Operators that
415 # need per-subscriber isolation should run one node per subscriber.
416 print(
417 json.dumps(
418 {
419 "stigmem_wake": {
420 "event_id": event["id"],
421 "subscription_id": event["subscription_id"],
422 "subscriber_identity": event["subscriber_identity"],
423 "delivery_address": event["delivery_address"],
424 "event_type": event["event_type"],
425 "fact": sanitized,
426 "ts": datetime.now(UTC).isoformat(),
427 }
428 }
429 ),
430 file=sys.stderr,
431 )
432 return True
435def _record_result(event: Any, success: bool) -> None:
436 now = datetime.now(UTC).isoformat()
437 new_attempts = (event["delivery_attempts"] or 0) + 1
439 if success:
440 with db() as conn:
441 conn.execute(
442 """UPDATE subscription_events
443 SET delivered_at=?, delivery_status='delivered',
444 delivery_attempts=?, claimed_at=NULL
445 WHERE id=?""",
446 (now, new_attempts, event["id"]),
447 )
448 conn.execute(
449 "UPDATE subscriptions SET last_delivered_at=?, consecutive_failures=0 WHERE id=?",
450 (now, event["subscription_id"]),
451 )
452 return
454 backoff_s = min(2**new_attempts, 300)
455 next_retry = datetime.fromtimestamp(time.time() + backoff_s, UTC).isoformat()
457 with db() as conn:
458 # Release the claim back to 'pending' so the next sweep can retry it.
459 # Circuit-breaker logic below may override to 'failed'.
460 conn.execute(
461 """UPDATE subscription_events
462 SET delivery_status='pending', claimed_at=NULL,
463 delivery_attempts=?, next_retry_at=?
464 WHERE id=?""",
465 (new_attempts, next_retry, event["id"]),
466 )
467 conn.execute(
468 "UPDATE subscriptions SET consecutive_failures=consecutive_failures+1 WHERE id=?",
469 (event["subscription_id"],),
470 )
471 sub = conn.execute(
472 "SELECT consecutive_failures FROM subscriptions WHERE id=?",
473 (event["subscription_id"],),
474 ).fetchone()
475 threshold = _settings_pkg.settings.subscription_circuit_threshold
476 if sub and sub["consecutive_failures"] >= threshold:
477 conn.execute(
478 "UPDATE subscriptions SET circuit_open=1 WHERE id=?",
479 (event["subscription_id"],),
480 )
481 conn.execute(
482 "UPDATE subscription_events SET delivery_status='failed' WHERE id=?",
483 (event["id"],),
484 )
485 logger.warning("Circuit breaker opened for subscription %s", event["subscription_id"])
488def _mark_delivered(event_id: str, subscription_id: str) -> None:
489 now = datetime.now(UTC).isoformat()
490 with db() as conn:
491 conn.execute(
492 """UPDATE subscription_events
493 SET delivered_at=?,
494 delivery_status='delivered',
495 delivery_attempts=delivery_attempts+1,
496 claimed_at=NULL
497 WHERE id=?""",
498 (now, event_id),
499 )
500 conn.execute(
501 "UPDATE subscriptions SET last_delivered_at=?, consecutive_failures=0 WHERE id=?",
502 (now, subscription_id),
503 )
506# ---------------------------------------------------------------------------
507# Background sweep loop
508# ---------------------------------------------------------------------------
511async def sweep_loop() -> None:
512 """Long-running asyncio task: runs deliver_pending() every N seconds."""
513 while True:
514 await asyncio.sleep(_settings_pkg.settings.subscription_delivery_sweep_s)
515 try:
516 await asyncio.to_thread(deliver_pending)
517 except Exception as exc:
518 logger.exception("subscription delivery sweep failed: %s", exc)