Cache results in the metagraph server

[karasikov]

Add /search result cache for the metagraph server

Why

Two real-world failure modes today:

1. Concurrent identical requests recompute independently. When two clients submit the same /search payload, both threads run the full multi-second-to-multi-minute query.
2. A connection drop during a long query forces a full retry. If the upstream API retries after a TCP failure, the server happily redoes the whole computation.

This PR adds a result cache that solves both. A refcount-pinned design ensures an entry is never evicted while a thread is still computing or reading it.

What

• ServerQueryCache (src/cli/server_cache.{hpp,cpp}):
  • In-flight dedup via std::shared_future<ResultPtr>: the first arrival owns the promise, late arrivals attach to the same future.
  • Byte-budgeted LRU: total response size capped by --cache-size (GB). Entries with active waiters (currently being computed or read) are never evicted.
  • Two-class retention with priority for entries whose delivery failed:
    • DELIVERED — successfully delivered at least once. Sink state, normal LRU.
    • PROTECTED — delivery failed, kept with priority over DELIVERED for a sliding 2 h window. Every cache hit refreshes the window (the hit is itself evidence the upstream is still retrying). After 2 h with no hits, the entry graduates to the main cache (state stays PROTECTED but it loses priority and ages out via LRU). A successful retry flips it to DELIVERED.
    • Pass 2 of the eviction sweep falls back to LRU on PROTECTED entries when the cache is otherwise full of them, so memory stays bounded.
• Cache key — canonicalized request inputs (FASTA hash + query_mode, discovery_fraction, min_exact_match, max_num_nodes_per_seq_char, top_labels, align, plus graph identity). Per-request overrides are read directly from the JSON, mirroring process_search_request's resolution. Formatting flags (verbose_output, Accept-Encoding) are deliberately excluded.
• /search handler — acquires a Handle, computes on miss, reuses on hit. The on_sent async-write callback drives mark_delivered / mark_protected. Both single-graph and multi-graph paths share the cache layer.
• process_request — gains an optional on_sent(error_code) callback routed through the Simple-Web-Server send(callback) variant.
• CLI — --cache-size <GB> (default 1; 0 disables). Stored as double so 0.5 etc. work.
• One log line per cached request (only when cache enabled):
  • cache HIT 4.5 KB (occupancy 312.0/1024 MB, 47 entries)
  • cache STORE 12.5 KB (occupancy 324.0/1024 MB, 48 entries)
  • delivery FAILED (Connection reset by peer); cache entry protected for 120 min retry window

Mechanism and policy

Per-request flow

  ┌────────────────────────────────────────────────────────────────────┐
  │ POST /search                                                       │
  └─────────────────────────────┬──────────────────────────────────────┘
                                │
                       key = hash(FASTA) + mode + fractions + graph_id
                                │
                      ┌─────────▼─────────┐
                      │ cache.acquire(key)│  ── waiters++ ──┐
                      └─────────┬─────────┘                 │
                                │                           │
              ┌─────────────────┴────────────────┐          │
              ▼                                  ▼          │
        ┌──────────┐                       ┌─────────────┐  │
        │  MISS    │                       │    HIT      │  │
        │ (first)  │                       │ (in-flight  │  │
        └────┬─────┘                       │  or ready)  │  │
             │                             └──────┬──────┘  │
             │ compute query_fasta()              │         │
             │ (single- or multi-graph)           │         │
             ▼                                    │         │
       handle.set_result(json) ──┐                │         │
       or set_exception(eptr)    │                │         │
                                 ▼                ▼         │
                          ┌──────────────────────────┐      │
                          │  shared_future<ResultPtr>│      │
                          │   .get()  →  Json::Value │      │
                          └──────────────┬───────────┘      │
                                         │                  │
                                  write response            │
                                         │                  │
                            on_sent(ec) callback fires      │
                              │                  │          │
                              ▼                  ▼          │
                      mark_delivered      mark_protected    │
                              │                  │          │
                              └────────┬─────────┘          │
                                       ▼                    │
                                ~Handle()  ── waiters-- ────┘

Entry state machine

                  acquire (miss)
                        │
                        ▼
                   ┌──────────┐
                   │ PENDING  │      computation in flight
                   └─────┬────┘      waiters ≥ 1, never evicted
                         │ set_result(...) / set_exception(...)
                         ▼
                  (response ready)
                         │
                  on_sent(ec) fires
                         │
              ┌──────────┴──────────┐
       ec != 0│                     │ec == 0
              ▼                     ▼
       ┌─────────────┐        ┌──────────┐
       │  PROTECTED  │        │ DELIVERED│   sink state — a later
       │             │        │          │   mark_protected() is
       │ within 2h   │        └──────────┘   a no-op.
       │  of last    │                       Normal LRU.
       │   hit:      │
       │ priority    │       successful retry  ┌──────────┐
       │ over        │ ────────────────────►   │ DELIVERED│
       │ DELIVERED.  │                          └──────────┘
       │             │
       │ past 2h:    │
       │ graduates   │
       │ to main     │
       │ cache       │
       │ (LRU like   │
       │ any other)  │
       └─────────────┘

   each acquire() during PROTECTED refreshes ready_at = now (sliding window)

Eviction policy (two-pass size-pressure sweep)

                  Eviction priority (most → least evictable)
                  ─────────────────────────────────────────────
                    DELIVERED              (LRU oldest first)
                    PROTECTED past 2h      (LRU oldest first)
                  ─── pass 1 boundary ────────────────────────
                    PROTECTED within 2h    (LRU oldest first)   ← held with
                                                                  priority
                                                                  unless cache
                                                                  has only
                                                                  these left
                  ─── pass 2 boundary ────────────────────────
                    waiters > 0            (NEVER evicted)


  ┌────────────────────────────────────────────────────────────────┐
  │ Pass 1: while over budget,                                     │
  │   walk LRU back→front, evict the oldest waiterless entry that  │
  │   is NOT PROTECTED-within-window.                              │
  │                                                                │
  │ Pass 2: still over budget? Walk again, evict any waiterless    │
  │   entry. Bounds memory under heavy PROTECTED pressure.         │
  │                                                                │
  │ Stops when under cap, or no candidate remains.                 │
  │                                                                │
  │ Triggered on insert (set_result) and on waiter release.        │
  │                                                                │
  │ refcount invariant:                                            │
  │   ── producer & readers ⇒ waiters ≥ 1                          │
  │   ── only ~Handle() drops the count                            │
  │   ── in-flight / being-read entries are NEVER evicted          │
  └────────────────────────────────────────────────────────────────┘

Three retention classes

State	Window	Eviction priority
PENDING	none	never (waiters always ≥ 1)
DELIVERED	none	LRU, normal priority
PROTECTED within 2 h	sliding	held over DELIVERED (last resort)
PROTECTED past 2 h	window past	LRU, normal priority

Concurrent dedup (the shared_future hinge)

   Thread A                    Thread B (50 ms later)
   ─────────                   ──────────────────────
   acquire(k)  → MISS          ⋮
   compute_search(...)         acquire(k)  → HIT (same future)
        ⋮                      future.get()  ── BLOCKS ──┐
        ⋮ (12 sec)                                       │
   set_result(json)  ──────────────────────────► both unblock
                                                         │
   future.get() returns        future.get() returns ◄────┘
   (instant)                   (instant)

The cache stores a shared_future<shared_ptr<const CachedResult>>. The first arrival owns the promise; every later arrival just .get()s the future — no extra computation. Result is published once, observed N times.

What's in the cache key (and what isn't)

  IN  the key  ──►  FASTA-body hash, query_mode (resolved from JSON
                    flags: query_coords, query_counts, abundance_sum,
                    with_signature), discovery_fraction,
                    min_exact_match, max_num_nodes_per_seq_char,
                    top_labels, align, graph identity (path or
                    sorted multi-graph list).

  OUT of key  ──►  verbose_output      (server-startup constant)
                   Accept-Encoding     (compression is post-cache)
                   request_id          (different requests, same answer)

Cache stores semantic Json::Value; per-request flags that affect only formatting or transport are applied on the way out.

Out of scope (follow-ups)

• Application-level handshake. Today the protection trigger is TCP-level (the on_sent error_code). A future POST /ack {request_id} can drive the same mark_delivered / mark_protected hooks — the API is symmetric.
• Cache invalidation on graph reload. Graph identity is part of the key, so a swapped graph file just gets a different namespace; explicit invalidation will be added if/when graph hot-reload becomes a real workflow.

Test plan

• Unit tests (tests/cli/test_server_cache.cpp, 14 cases): miss→hit, concurrent dedup, LRU under size pressure, refcount protection, PROTECTED retention, sliding-window refresh on hit, graduation to main cache past window, DELIVERED-as-sink immunity to later mark_protected, PROTECTED-within-window held over DELIVERED, PROTECTED entries evicted under heavy pressure, disabled cache (--cache-size 0), exception propagation, abandoned-producer.
• All 3,653+ unit tests pass.
• All 171 test_api integration tests pass (verified all query_mode paths still distinguish correctly with the resolved-from-JSON cache key).
• All 3,594 integration tests pass.
• Manual integration test: hit /search twice with identical payload → second response is sub-second; kill curl mid-response on a long query → re-issue same request → cache hit.