src/monitor_app/cached_product.py

0001 """Uniform handling of long-build caching (docs/CACHED_PRODUCTS.md).
0002
0003 A cached product is any expensive-to-build, read-often result. The
0004 contract: a request always serves the stored product immediately,
0005 stamped with its build time — nothing expensive builds in the request
0006 path. Staleness triggers a background rebuild behind the response; an
0007 explicit update request rebuilds synchronously because the user chose
0008 to wait. ``building_since`` is the cross-worker in-flight lock, so
0009 concurrent requests never stampede a rebuild.
0010
0011 Pure-database builders run here in a background thread. Credentialed or
0012 very heavy builds belong on the prod-ops agent with an SSE completion
0013 push (EPICPROD_OPS_AGENT.md) — this module is the light half of the one
0014 pattern.
0015 """
0016
0017 import logging
0018 import threading
0019 import time
0020 from datetime import timedelta
0021
0022 from django.db import close_old_connections
0023 from django.utils import timezone
0024
0025 logger = logging.getLogger(__name__)
0026
0027 # A build slot older than this is considered abandoned (a worker died
0028 # mid-build) and may be reclaimed.
0029 BUILD_LOCK_TIMEOUT_SECONDS = 600
0030
0031 # An explicit update that finds another worker mid-build waits up to this
0032 # long for that build to land before giving up and serving what exists.
0033 REFRESH_WAIT_SECONDS = 15
0034
0035
0036 def _claim(key):
0037     """Atomically claim the build slot for a key. True when claimed."""
0038     from django.db.models import Q
0039
0040     from .models import CachedProduct
0041
0042     now = timezone.now()
0043     stale_lock = now - timedelta(seconds=BUILD_LOCK_TIMEOUT_SECONDS)
0044     CachedProduct.objects.get_or_create(key=key)
0045     claimed = (CachedProduct.objects
0046                .filter(key=key)
0047                .filter(Q(building_since__isnull=True)
0048                        | Q(building_since__lt=stale_lock))
0049                .update(building_since=now))
0050     return claimed == 1
0051
0052
0053 def _build_and_store(key, builder):
0054     """Run the builder and store its product; the lock always clears.
0055
0056     Every failure is logged with the key — a broken builder must surface
0057     in the log, never present as silently-stale data.
0058     """
0059     from .models import CachedProduct
0060
0061     started = time.monotonic()
0062     try:
0063         value = builder()
0064         CachedProduct.objects.filter(key=key).update(
0065             value=value,
0066             built_at=timezone.now(),
0067             build_seconds=round(time.monotonic() - started, 3),
0068             building_since=None,
0069         )
0070     except Exception:
0071         logger.exception('cached product build failed: %s', key)
0072         CachedProduct.objects.filter(key=key).update(building_since=None)
0073         raise
0074
0075
0076 def _background_build(key, builder):
0077     def run():
0078         close_old_connections()
0079         try:
0080             _build_and_store(key, builder)
0081         except Exception:
0082             pass  # already logged with the key in _build_and_store
0083         finally:
0084             close_old_connections()
0085
0086     thread = threading.Thread(
0087         target=run, name=f'cached-product-{key[:40]}', daemon=True)
0088     thread.start()
0089
0090
0091 def get_product(key, builder, ttl_seconds, refresh=False):
0092     """Serve a cached product; rebuild by the contract above.
0093
0094     Returns ``{'value', 'built_at', 'age_seconds', 'refreshing',
0095     'built_now'}``. The first-ever fill and an explicit ``refresh`` build
0096     synchronously (there is nothing to serve, or the user asked and
0097     waits); a stale product returns immediately while a background
0098     rebuild runs.
0099     """
0100     from .models import CachedProduct
0101
0102     row = CachedProduct.objects.filter(key=key).first()
0103     have_product = row is not None and row.built_at is not None
0104
0105     if (refresh or not have_product) and _claim(key):
0106         _build_and_store(key, builder)
0107         row = CachedProduct.objects.filter(key=key).first()
0108         return {
0109             'value': row.value,
0110             'built_at': row.built_at,
0111             'age_seconds': 0.0,
0112             'refreshing': False,
0113             'built_now': True,
0114         }
0115
0116     if refresh and have_product:
0117         # Explicit update, but another worker holds the build lock. The
0118         # caller chose to wait, so wait briefly for that build to land
0119         # rather than returning stale data as if it were the update.
0120         prior_built = row.built_at
0121         deadline = time.monotonic() + REFRESH_WAIT_SECONDS
0122         while time.monotonic() < deadline:
0123             time.sleep(0.5)
0124             row = CachedProduct.objects.filter(key=key).first()
0125             if row is not None and row.built_at is not None \
0126                     and row.built_at != prior_built:
0127                 return {
0128                     'value': row.value,
0129                     'built_at': row.built_at,
0130                     'age_seconds': round(
0131                         (timezone.now() - row.built_at).total_seconds(), 1),
0132                     'refreshing': False,
0133                     'built_now': False,
0134                 }
0135             if (row is None or row.building_since is None) and _claim(key):
0136                 # The other build ended without landing a newer product
0137                 # (failure or lock expiry). The caller asked for a
0138                 # synchronous update, so build it here after all.
0139                 _build_and_store(key, builder)
0140                 row = CachedProduct.objects.filter(key=key).first()
0141                 return {
0142                     'value': row.value,
0143                     'built_at': row.built_at,
0144                     'age_seconds': 0.0,
0145                     'refreshing': False,
0146                     'built_now': True,
0147                 }
0148         # The in-flight build outlasted the wait; fall through and report
0149         # the stored product with refreshing status honestly.
0150
0151     if row is None or row.built_at is None:
0152         # Another worker holds the first-fill lock; nothing to serve yet.
0153         return {'value': None, 'built_at': None, 'age_seconds': None,
0154                 'refreshing': True, 'built_now': False}
0155
0156     age = (timezone.now() - row.built_at).total_seconds()
0157     refreshing = row.building_since is not None
0158     if age > ttl_seconds and not refreshing and _claim(key):
0159         _background_build(key, builder)
0160         refreshing = True
0161     return {
0162         'value': row.value,
0163         'built_at': row.built_at,
0164         'age_seconds': round(age, 1),
0165         'refreshing': refreshing,
0166         'built_now': False,
0167     }