feat: the membership sweep, and the state that makes its failures readable (387 C3)
CI / lint (push) Failing after 3s
Build images / sign-extension (push) Successful in 4s
CI / extension-version (push) Successful in 2s
Build images / build-agent (push) Successful in 8s
CI / frontend-build (push) Successful in 28s
CI / backend-lint-and-test (push) Successful in 38s
Build images / build-web (push) Successful in 1m28s
Build images / smoke-web (push) Skipped
Build images / build-ml (push) Successful in 2m35s
Build images / promote (push) Skipped
CI / integration (push) Successful in 2m55s
CI / lint (push) Failing after 3s
Build images / sign-extension (push) Successful in 4s
CI / extension-version (push) Successful in 2s
Build images / build-agent (push) Successful in 8s
CI / frontend-build (push) Successful in 28s
CI / backend-lint-and-test (push) Successful in 38s
Build images / build-web (push) Successful in 1m28s
Build images / smoke-web (push) Skipped
Build images / build-ml (push) Successful in 2m35s
Build images / promote (push) Skipped
CI / integration (push) Successful in 2m55s
A daily sweep that walks each platform's roster into `platform_membership`. Daily because memberships change on a BILLING cycle, not a download cadence. **`membership_sync` is the part that earns its keep.** Without it three very different situations are one indistinguishable state — the account subscribes to nothing, the sweep never ran, the sweep failed — and all three leave zero rows in `platform_membership`. "You are tracking 12 sources you no longer subscribe to" is correct in the first case and an invitation to cancel things the operator is actively paying for in the other two. So C4 gates its CONCLUSIONS on `last_success_at`, not merely its display, and `roster_is_fresh` is computed server-side so no caller can forget to. Two timestamps rather than one: `last_attempt_at` moves every run, `last_success_at` only on a clean walk. The gap between them is the signal — a sweep hammering a broken credential every day must not look healthy because it ran recently, and there is a test for exactly that. Rejected shortcuts, both tempting: `MAX(platform_membership.last_seen_at)` cannot tell "synced fine, found nothing" from "never synced"; `task_run` is worse, since its retention prunes ok rows after 24h and a sweep that last succeeded three days ago would leave no trace at all. **The fetch completes before anything is written.** That ordering is the safety property: a walk that dies mid-pagination writes nothing, so a failure can never leave a roster half this week's and half last week's. `touch_membership` never deletes, so a failure cannot empty the roster either — but "intact" should mean intact, not merely non-empty. Rule 89's four, each where it actually lives: recovery is "run it again" (upsert, no deletes); retention is C1's age-out-never-delete, because disappearing IS the signal; the wall-clock deadline is per-platform and distinct from the per-REQUEST timeout the client already has (rule 156 — a paginated roster answering every page slowly-but-within-timeout would never trip that one and would sit on a worker indefinitely); duration comes from the existing TaskRun signal plumbing. **A bug caught in review, not production:** the broad `except Exception` would have swallowed Celery's SoftTimeLimitExceeded — which is an ORDINARY Exception subclass, not a BaseException — letting the sweep run past the soft limit into the hard one, where it is SIGKILLed mid-transaction. A sweep that cannot be stopped is worse than one that fails. Now re-raised explicitly, with a test that also asserts SoftTimeLimitExceeded is still an Exception, so the re-raise cannot quietly become dead code. Rule 164 is why this ships with UI rather than backend-only: a roster that never synced must be VISIBLE as such. The card says "never synced" in words and states no count at all — rendering it as 0 is the precise conflation the whole step exists to prevent — while a real zero behind a real sync is reported as zero, because that one IS an answer. Pinned in both directions. Three independent gates decide whether a platform is swept — registered here, client exposes `iter_memberships`, credential exists — each silent, so adding SubscribeStar (D1) is one line and nothing else. A missing credential is not an error: recording a failure would light up the UI for a feature never enabled. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LNXXULQDjVZmbuNa2G9mD9
This commit is contained in:
@@ -28,11 +28,14 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from sqlalchemy import func
|
||||
from collections.abc import Awaitable, Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from ..models import PlatformMembership
|
||||
from ..models import MembershipSync, PlatformMembership
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -146,3 +149,130 @@ async def touch_membership(
|
||||
},
|
||||
)
|
||||
await session.execute(stmt)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The sweep, and the state that makes its failures readable (#387 C3)
|
||||
# ---------------------------------------------------------------------------
|
||||
#
|
||||
# How long a successful sync stays trustworthy. Beyond this the roster is
|
||||
# STALE, and C4 must refuse to draw conclusions from it — "you are tracking 12
|
||||
# sources you no longer subscribe to", computed from a roster that stopped
|
||||
# syncing a week ago, is an invitation to cancel things the operator is still
|
||||
# paying for.
|
||||
#
|
||||
# Generous relative to the daily cadence: a few missed runs are a blip, not a
|
||||
# reason to stop trusting a roster that changes on a billing cycle.
|
||||
ROSTER_STALE_AFTER = timedelta(days=3)
|
||||
|
||||
|
||||
async def get_sync_state(session: AsyncSession, platform: str) -> MembershipSync | None:
|
||||
return (await session.execute(
|
||||
select(MembershipSync).where(MembershipSync.platform == platform)
|
||||
)).scalar_one_or_none()
|
||||
|
||||
|
||||
def roster_is_fresh(state: MembershipSync | None, *, now: datetime | None = None) -> bool:
|
||||
"""May a caller draw CONCLUSIONS from this roster?
|
||||
|
||||
False for never-synced and for stale, and those are deliberately the same
|
||||
answer here even though the UI must tell them apart: both mean the roster
|
||||
is not evidence. The asymmetry that matters is that `False` never means
|
||||
"you subscribe to nothing" — it means "we do not know", and a caller that
|
||||
cannot represent "we do not know" must not be asking this question.
|
||||
"""
|
||||
if state is None or state.last_success_at is None:
|
||||
return False
|
||||
now = now or datetime.now(UTC)
|
||||
return (now - state.last_success_at) <= ROSTER_STALE_AFTER
|
||||
|
||||
|
||||
async def _record_sync(session: AsyncSession, platform: str, **values) -> None:
|
||||
stmt = pg_insert(MembershipSync).values(platform=platform, **values)
|
||||
await session.execute(stmt.on_conflict_do_update(
|
||||
constraint="uq_membership_sync_platform",
|
||||
set_={**values, "updated_at": func.now()},
|
||||
))
|
||||
|
||||
|
||||
async def sync_platform(
|
||||
session: AsyncSession,
|
||||
*,
|
||||
platform: str,
|
||||
fetch: Callable[[], Awaitable[list]],
|
||||
now: datetime | None = None,
|
||||
) -> dict:
|
||||
"""Walk one platform's roster and record what happened.
|
||||
|
||||
`fetch` is injected rather than built here so the error-to-state mapping —
|
||||
the part with the consequences — is testable without a credential, and so
|
||||
this service needs to know nothing about how any particular client is
|
||||
constructed.
|
||||
|
||||
THE FETCH COMPLETES BEFORE ANYTHING IS WRITTEN. That ordering is the whole
|
||||
safety property: a walk that dies half way through pagination writes
|
||||
nothing, so a failure can never leave a roster that is partly this week's
|
||||
and partly last week's. (`touch_membership` never deletes, so a failure
|
||||
cannot empty the roster either — but "intact" should mean intact, not
|
||||
merely non-empty.)
|
||||
|
||||
Returns a summary dict; never raises for a platform failure, because one
|
||||
platform failing must not abort the others.
|
||||
"""
|
||||
now = now or datetime.now(UTC)
|
||||
await _record_sync(session, platform, last_attempt_at=now)
|
||||
await session.commit()
|
||||
|
||||
try:
|
||||
memberships = await fetch()
|
||||
except Exception as exc: # noqa: BLE001 - deliberately broad, see below
|
||||
# Broad on purpose: a sweep is a background job, and ANY escape here
|
||||
# kills the run for every other platform too. The exception's class
|
||||
# name is recorded so the distinction the client drew (auth vs drift
|
||||
# vs transport) survives into the UI, which is where it is actionable.
|
||||
#
|
||||
# EXCEPT the worker asking us to stop. Celery raises its soft time
|
||||
# limit as an ordinary Exception subclass, so a broad catch swallows
|
||||
# the shutdown request and lets the sweep run on into the HARD limit,
|
||||
# where it is SIGKILLed mid-transaction. A sweep that cannot be stopped
|
||||
# is worse than one that fails. (KeyboardInterrupt and SystemExit are
|
||||
# BaseException and pass through this clause already.)
|
||||
from celery.exceptions import SoftTimeLimitExceeded
|
||||
|
||||
if isinstance(exc, SoftTimeLimitExceeded):
|
||||
raise
|
||||
await session.rollback()
|
||||
await _record_sync(
|
||||
session, platform,
|
||||
last_error_type=type(exc).__name__,
|
||||
last_error_message=str(exc)[:2000],
|
||||
)
|
||||
await session.commit()
|
||||
log.warning("membership sync failed for %s: %s", platform, exc)
|
||||
return {"platform": platform, "ok": False, "error": type(exc).__name__}
|
||||
|
||||
for m in memberships:
|
||||
await touch_membership(
|
||||
session,
|
||||
platform=platform,
|
||||
external_campaign_id=m.campaign_id,
|
||||
display_name=m.display_name,
|
||||
url=m.url,
|
||||
status=m.status,
|
||||
tier_names=m.tier_names or None,
|
||||
amount_cents=m.amount_cents,
|
||||
currency=m.currency,
|
||||
details={**(m.details or {}), "is_free_member": m.is_free_member},
|
||||
)
|
||||
await _record_sync(
|
||||
session, platform,
|
||||
last_success_at=now,
|
||||
last_count=len(memberships),
|
||||
# Cleared on success — a stale error beside a fresh success would read
|
||||
# as "still broken" forever.
|
||||
last_error_type=None,
|
||||
last_error_message=None,
|
||||
)
|
||||
await session.commit()
|
||||
log.info("membership sync ok for %s: %d membership(s)", platform, len(memberships))
|
||||
return {"platform": platform, "ok": True, "count": len(memberships)}
|
||||
|
||||
Reference in New Issue
Block a user