feat(consumer): bound Shape.Consumer heap growth via spawn opts + adaptive GC

[erik-the-implementer]

Closes #4476

Summary

Bounds runaway old-heap growth of Shape.Consumer processes (issue #4476). On a production node with 5,043 consumers, ~36 GB was unreclaimable garbage (7–9 MB heaps each holding ~8 KB of live state) because consumers inherit the BEAM default fullsweep_after: 65535 and rarely hibernate, so a full sweep is effectively never triggered.

Two complementary, independently-tunable levers:

1. Per-process spawn opts (env-driven). Consumer, Consumer.Snapshotter, and Consumer.Materializer now pass spawn_opt: Electric.StackConfig.spawn_opts(stack_id, key) (keys :consumer, :consumer_snapshotter, :consumer_materializer), mirroring ShapeLogCollector. This lets ELECTRIC_PROCESS_SPAWN_OPTS set fullsweep_after per process type. No baked-in defaults — purely env-configured.

2. Adaptive GC, opt-in. After processing a transaction fragment (including during the startup buffer drain), the consumer forces :erlang.garbage_collect() only if its heap exceeds consumer_gc_heap_threshold (bytes; nil = off, the default). The threshold is cached per-consumer at startup (State.new/2), so changing it affects consumers started afterwards. The forced sweep runs off the reply path via a {:continue, :maybe_gc} step, so it never blocks the ShapeLogCollector's synchronous publish call, and a hysteresis interval (@gc_min_interval_ms) caps how much CPU a busy consumer spends sweeping. See Electric.Shapes.Consumer.set_gc_heap_threshold/2.

Request-handler (GET /v1/shape) processes are intentionally untouched — they already expose fullsweep_after via ELECTRIC_TWEAKS_HANDLER_FULLSWEEP_AFTER and force GC before long-poll blocking.

Config

Env var	Meaning	Default
ELECTRIC_PROCESS_SPAWN_OPTS	JSON map; now accepts consumer/consumer_snapshotter/consumer_materializer keys (e.g. {"consumer":{"fullsweep_after":4}})	%{}
ELECTRIC_CONSUMER_GC_HEAP_THRESHOLD	adaptive-GC heap threshold in bytes; nil disables	nil

🤖 Generated with Claude Code

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 58.74%. Comparing base (176bec8) to head (bb6425d).
⚠️ Report is 8 commits behind head on main.
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4539      +/-   ##
==========================================
+ Coverage   58.35%   58.74%   +0.38%     
==========================================
  Files         370      383      +13     
  Lines       40674    42260    +1586     
  Branches    11553    12103     +550     
==========================================
+ Hits        23735    24824    +1089     
- Misses      16865    17360     +495     
- Partials       74       76       +2     

Flag	Coverage Δ
packages/agents	72.43% <ø> (+1.05%)	⬆️
packages/agents-mcp	77.70% <ø> (ø)
packages/agents-mobile	78.81% <ø> (+3.32%)	⬆️
packages/agents-runtime	82.57% <ø> (+0.10%)	⬆️
packages/agents-server	74.92% <ø> (+0.06%)	⬆️
packages/agents-server-ui	7.36% <ø> (+1.10%)	⬆️
packages/electric-ax	46.42% <ø> (ø)
packages/experimental	87.73% <ø> (ø)
packages/react-hooks	86.48% <ø> (ø)
packages/start	82.83% <ø> (ø)
packages/typescript-client	91.83% <ø> (ø)
packages/y-electric	56.05% <ø> (ø)
typescript	58.74% <ø> (+0.38%)	⬆️
unit-tests	58.74% <ø> (+0.38%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[claude]

Claude Code Review

Summary

Bounds runaway old-heap growth in Shape.Consumer / Snapshotter / Materializer (issue #4476) via two inert-by-default levers: per-process spawn_opt (ELECTRIC_PROCESS_SPAWN_OPTS) and an opt-in adaptive GC that fires once a consumer's heap exceeds ELECTRIC_CONSUMER_GC_HEAP_THRESHOLD. Iteration 5: the adaptive GC is now run off the reply path via {:continue, :maybe_gc} (b06599795) so a forced full sweep no longer blocks the ShapeLogCollector's synchronous publish, and the GC integration tests now assert on State.last_forced_gc_at instead of fragile heap-size / GC-trace magnitudes. The new code is clean and correct. The one carry-over remains the stale PR description.

What is Working Well

• Deferring the GC off the reply path is a real, well-reasoned improvement. The SLC fans events out to consumers via $gen_call and collects each :ok before the fragment is considered published (shape_log_collector.ex:123). Previously the forced :erlang.garbage_collect() ran inside handle_event before the reply, so the SLC's reply-collection blocked on each consumer's sweep. Now handle_call({:handle_event, …}) returns {:reply, :ok, state, {:continue, :maybe_gc}} (consumer.ex:237) — the SLC gets :ok immediately and can move on while this consumer sweeps. Because handle_continue is guaranteed to run before any mailbox message is dequeued, the GC still completes before the next fragment is handled, so there's no continue pile-up and no change to per-consumer ordering. Client and materializer notifications already happened before the reply, so the deferral doesn't delay them either.
• hibernate_after is correctly re-armed. A {:continue, …} return can't carry the inactivity timeout, and handle_continue(:maybe_gc, …) restores it ({:noreply, state, state.hibernate_after}, consumer.ex:202-205). Both the live path and the :consume_buffer drain route through it, so the timeout is preserved on every event path.
• The test rewrite is a genuine robustness upgrade. Asserting on last_forced_gc_at (the consumer's own forced-GC decision, stamped iff it actually swept) is immune to off-heap binaries and incidental BEAM major GCs that made the prior heap-size and gc_major_start-trace assertions fragile. The flush mechanism is sound: :sys.get_state/1 is a system message that queues behind the pending :maybe_gc continue, so it can only observe post-GC state — no sleeps, no races.
• Carries over from earlier iterations: the set_gc_heap_threshold_all_stacks/1 async-test concern is gone; the threshold is cached at startup into State.gc_heap_threshold so the hot-path check is a struct-field match with a nil fast path; should_force_gc?/5 remains a clean, injectable, boundary-tested pure predicate; changeset present and accurate.

Issues Found

Critical (Must Fix): None.

Important (Should Fix):

• PR description is still stale and now contradicts the code in two more places (carry-over from iteration 4, still unaddressed). The "Adaptive GC, opt-in, runtime-tunable" section still says "The threshold is read from StackConfig on every fragment, so it can be changed live from IEx — see Electric.Shapes.Consumer.set_gc_heap_threshold/2 and set_gc_heap_threshold_all_stacks/1." After the cache-at-startup refactor (f8e757891) and the all-stacks helper removal, both halves are wrong: the threshold is read once at consumer startup in State.new/2 (so set_gc_heap_threshold/2 only affects consumers started after the call — already-running consumers keep their boot-time value), and set_gc_heap_threshold_all_stacks/1 no longer exists. The Rollout/rollback line "The threshold can be toggled live from IEx without a deploy" has the same problem. Additionally, the description still says the GC runs synchronously "after processing a transaction fragment" and makes no mention of the {:continue, :maybe_gc} deferral introduced this iteration. The in-code docstring on set_gc_heap_threshold/2 (consumer.ex:119-124) is correct — please bring the PR body in line so an operator following it during an incident isn't misled.

Suggestions (Nice to Have):

• The buffer drain now defers a single GC to the end instead of one (rate-limited) GC per fragment. process_buffered_txn_fragments/1 no longer calls maybe_garbage_collect/1 per fragment; the :consume_buffer continue defers one GC once the whole buffer is drained (consumer.ex:188-196, 883-892). On the reply path this is the right call. But the buffer drain is triggered by the :pg_snapshot_known cast, not a synchronous SLC call — there's no caller being unblocked there — so for a large startup buffer the only effect is that intermediate sweeps are dropped and peak heap during the drain can be higher than in iteration 4 (which swept up to once/sec mid-drain). Since the drained fragments were already buffered in memory and this is precisely the accumulation scenario Shape consumers accumulate multi-MB heaps of unreclaimed floating garbage (default fullsweep_after, rarely hibernate) #4476 targets, it's worth confirming this is an acceptable trade vs. keeping a rate-limited mid-drain sweep. Low priority — the steady-state path is the one that matters most.
• Still no observability when the forced GC fires (carry-over). maybe_garbage_collect/1 calls :erlang.garbage_collect() and stamps last_forced_gc_at with no Logger line or :telemetry.execute/3 (consumer.ex:562-573). Rate-limited to ≤1/sec/consumer, a single throttled counter/event would be cheap and would let an operator confirm the backstop is firing — and aligns with the project's "new features emit telemetry" convention. The tests now reading last_forced_gc_at confirm this is the natural signal to surface. Minor; opt-in feature.
• The floating comment block at consumer.ex:553-557 documenting maybe_garbage_collect/1's calling convention is separated from the function by a blank line and the # Fast path: comment, so it reads as orphaned. Consider attaching it directly to the first maybe_garbage_collect/1 clause. Trivial.
• Carry-overs (by design, unchanged): @gc_min_interval_ms is compile-time only while the threshold is runtime-config; the heap bound is approximately threshold + a drain's worth of allocations rather than a hard cap (documented latency-vs-heap trade-off); a negative ELECTRIC_CONSUMER_GC_HEAP_THRESHOLD passes :integer env parsing and is only rejected by the NimbleOptions schema at stack init (fail-fast, acceptable).

Issue Conformance

Still matches #4476 precisely — the spawn-opt lever targets exactly the three named processes (Consumer, Snapshotter, Materializer), request handlers are correctly left untouched (they already have handler_fullsweep_after), and the adaptive GC is the backstop. Closes #4476, changeset present, no scope creep. The only conformance gap is the PR description drift noted above — the issue itself is well-specified.

Previous Review Status

Iteration 4 → 5.

• Resolved: nothing regressed; the GC integration tests were made less fragile (last_forced_gc_at instead of heap magnitude / GC traces).
• New this pass: the {:continue, :maybe_gc} deferral (b06599795) moves the forced sweep off the SLC reply path — reviewed and correct, with hibernate_after properly re-armed. Introduced one new (low-priority) trade-off: the buffer drain now GCs once at the end rather than mid-drain.
• Still open: the PR description (Important) — flagged in iteration 4, not yet updated, and now further out of sync after the deferral. No telemetry on forced GC (Suggestion).

The code is in good shape. The single actionable item remains updating the PR description/runbook wording to match the cached-at-startup + deferred-GC behavior.

---

Review iteration: 5 | 2026-06-15

[robacourt]

🚀

[github-actions]

This PR has been released! 🚀

The following packages include changes from this PR:

• @core/sync-service@1.7.1

Thanks for contributing to Electric!