3.76 GB allocated against 11 MB reachable, growing ~10 MB/min; .sys.gc reclaims nothing

[vbmithr]

Summary

A long-lived Rayforce service reports 3.76 GB allocated against 11.4 MB of
reachable data, growing ~10 MB/min indefinitely. .sys.gc reclaims 32 KB of
it. A second service on the same build, same IPC subscription path and a
larger working set is clean, so this is not simply how the allocator behaves.

I could not reduce it to a primitive. What follows is the evidence and
everything I ruled out, in case it is enough for you to recognise.

The process

An incremental consumer: subscribes to a multicast topic, buffers arriving
rows in a global table, and every 5 s runs a pass of group-bys, joins and
anti-joins over what arrived, appending results to a findings table that is
trimmed to a 100k-row cap.

(.sys.mem)
  bytes-allocated : 3,764,364,736     (3.76 GB)
  peak-bytes      : 3,787,658,624
  alloc-count     : 69,584,761
  sys-current     : 4,330,172,416
  anon-committed  : 4,328,521,728

(.mem.objsize findings) :  5,374,456   (5.4 MB,  94k rows)
(.mem.objsize depth)    :  5,967,396   (6.0 MB,  ~10k rows, oscillating)
(.mem.objsize last_msg) :     38,374   (38 KB,   774 rows)
(.sys.env)              :        407   globals

11.4 MB reachable through globals; 3.76 GB allocated. RSS 4.1 GB. Uptime 7 h.

It is retained, not uncollected

before (.sys.gc) : 3,766,785,344
after  (.sys.gc) : 3,766,752,576      (32,768 bytes reclaimed)

It grows linearly, while the working set does not

Sampled 60 s apart:

bytes-allocated	alloc-count	findings rows	depth rows
3,763,437,824	69,648,121	94,280	11,355
3,772,720,320	69,819,874	94,451	7,944
3,783,627,264	70,004,906	94,769	9,501

~10 MB/min, ~3,000 allocations/s, with both tables flat. That rate over the
uptime accounts for the whole 4.1 GB.

A control on the same build

Another service of ours, same Rayforce, also a multicast subscriber, with a
much larger working set (534k-row table), 98 min uptime:

bytes-allocated : 124 MB
(.mem.objsize levels) : 44 MB
(.mem.objsize output) : 50 MB

124 MB against 94 MB visible. No gap, no growth. So the IPC/subscribe path and
table mutation in general are not implicated.

Ruled out

Each run in a loop against tables of the shape the service uses (50k rows,
773 distinct symbol keys), with .sys.gc before and after, measuring
bytes-allocated delta. All reported 0 MB:

• anti-join on a two-column key, 200 reps
• inner-join on a two-column key, 200 reps
• select {by: ...} group-by, 200 reps
• select {where: ...}, 200 reps
• xasc, 100 reps
• concat, 200 reps
• insert into a global then trim with a negative take, 300 reps
• reassigning a global from an anti-join, 300 reps
• delete {from: 'global where: ...}, 300 reps
• format with 3 and 7 arguments, 20,000 reps
• map over til, 20,000 reps
• at on a column, 20,000 reps

So the leak is not in those operations standalone. What the real service adds
over the probes is that it runs them from a timer callback, on data arriving
through a multicast subscription, continuously for hours -- rather than in a
tight loop in one evaluation.

What would help

If there is a counter or a debug build that attributes live allocations by
call site, I am happy to run it against this process and report back -- it is
reproducible simply by leaving the service up. Equally, if alloc-count minus
frees is tracked anywhere, that delta would localise it quickly.

---

Measured on rayforce 2.6.2.r229.ge7770412.

[vbmithr]

Narrowed this considerably. It is not the query pass, the timer, the
ingest logic, or any primitive — each ruled out by direct measurement below.
What is left is that the leak only appears when the ingest runs on
IPC-delivered batches, and a structurally similar process on the same
topic and the same message rate does not leak at all.

The pass is clean

Measured in-process on the live service with .mem.ts:

(.mem.ts (run_checks))     -> net=0 KB  alloc=23,224 KB  time=4 ms
(.mem.ts (run_checks))     -> net=0 KB  alloc= 3,035 KB  time=0 ms
(.mem.ts (run_checks))     -> net=0 KB  alloc= 3,036 KB  time=0 ms
(.mem.ts (message_groups)) -> net=0 KB  alloc=   433 KB  time=0 ms

Net zero every time, while the process as a whole grows ~11 MB/min.

The ingest logic is clean when driven locally

I took the service's source, removed the subscription and the timer, and
drove its upd handler by hand with locally-built batches of the same
schema — 2,000 batches of 500 rows:

batches=  200  depth_rows= 100,500  delta=12 MB
batches=  600  depth_rows= 300,500  delta=49 MB
batches= 1000  depth_rows= 500,500  delta=49 MB
batches= 1400  depth_rows= 700,500  delta=52 MB
batches= 2000  depth_rows=1,000,000  delta=52 MB

52 MB for 1M retained rows is the data itself (~52 B/row, 7 columns) and it
stops growing once the row count does. No leak on this path.

Timer callbacks are clean

A repeating 100 ms timer whose callback runs a group-by, an anti-join and an
inner-join over a 50k-row table, 850 firings over 85 s: bytes-allocated
delta flat at 4 MB throughout.

A control that should leak and does not

Another of our services, same build, subscribed to the same topic on the
same tickerplant, receiving the same ~27 messages/s, 2 h uptime. Over 120 s:

bytes-allocated  133,127,552 -> 134,719,936   (+1.6 MB)
levels rows          611,536 ->     623,171   (+11,635)
held rows                  0 ->       2,094

+1.6 MB fully accounted for by +11,635 rows at ~82 B/row. RSS 322 MB, flat.
So IPC delivery by itself is not enough to trigger it.

The leaking process, same window

bytes-allocated  4,033,749,440 -> 4,056,388,736   (+22,639,296)
findings rows          100,000 ->       100,000   (at its cap, flat)
depth rows              42,240 ->        36,705   (oscillating, no trend)

+22.6 MB with zero net change in reachable state. Against 3,274 messages
delivered in that window that is ~6.9 KB per message; against 2,971,548
rows it is ~7.6 B/row. I cannot separate those two from a single operating
point — the feed does not offer me a second rows-per-message ratio.

So

The difference between the two processes is what they do with the delivered
table. The leaking one projects columns out of it ((get b 'ts_recv) and
friends) into a new table that it retains across passes; the clean one
builds a similar projection but drains and replaces its buffer within the
same batch. If a table deserialised from IPC can be kept alive by a
projection of its columns outliving the callback, that would fit every
measurement here — including why driving the identical code with
locally-built tables shows nothing.

That is a hypothesis, not a finding; I have no visibility into your
deserialisation path. But it is the shape the evidence points at, and it
would be cheap for you to confirm or dismiss from the inside.

[vbmithr]

Retracting the hypothesis in my previous comment. I tried to reproduce it
and could not; the attempt argues against the explanation I offered, so please
do not spend time on it.

I suggested that a table deserialised from IPC might be kept alive by a
projection of its columns outliving the callback. Two standalone reproducers
say no.

1. Request/response. A server holding a global, and a handler that
projects the delivered table's columns into a new table and appends it to that
global, trimmed to a 200k cap. A client sends 4,000 tables of 500 rows.
Against a control handler doing the identical projection and discarding it:

DROP (project, discard) :  0 MB after 4000 messages  acc=0
KEEP (project, retain)  :  8 MB after 4000 messages  acc=200,000
DROP again              :  0 MB after 4000 messages  acc=200,000
KEEP again              :  0 MB after 4000 messages  acc=200,000

The 8 MB is acc filling to its cap — 200k rows of four columns. The second
KEEP run, with acc already at the cap, costs nothing. No leak.

2. Multicast, the shape the real service actually uses. A publisher doing
.mc.pub of ~900-row batches every 37 ms (~27/s, matching our live rate), and
a subscriber whose upd projects and retains exactly as above:

 8 MB  acc=200,000  msgs=  334
 8 MB  acc=200,000  msgs= 1,550
 8 MB  acc=200,000  msgs= 2,767
 8 MB  acc=200,000  msgs= 3,983
 8 MB  acc=200,000  msgs= 5,199
 8 MB  acc=200,000  msgs= 6,416

Flat across 6,416 delivered messages. No leak.

So .mc.sub delivery, projection of a delivered table, retention across
callbacks, and the trim pattern are all clean in isolation and in combination.

Where that leaves it

The live service still grows ~11 MB/min with reachable state flat, and I still
cannot produce a minimal case. Everything I have ruled out, cumulatively:

• every query primitive it uses, standalone
• insert + negative-take trim; reassigning a global from an anti-join;
  delete on a global
• format, map, at
• work inside a repeating timer callback
• its own pass measured in-process: (.mem.ts (run_checks)) -> net 0 KB
• its ingest logic driven by hand with locally-built batches
• request/response IPC delivery with retention (above)
• multicast delivery with retention (above)
• a control service on the same build and the same topic at the same message
  rate, which does not leak at all

What my reproducers do not have, against the real service: a wider wire
schema, and concurrent ingest while a 5 s timer runs group-bys and joins over
the accumulated buffer and appends to a second, long-lived table. I have not
found a way to make those matter, but that is the remaining gap rather than a
claim.

I am happy to run anything against the live process — it reproduces simply by
being left up — or to try a build with allocation-site attribution if one
exists. I would rather hand you a minimal case, and I do not have one.

[vbmithr]

Found the allocation site. massif --pages-as-heap=yes sees it; every
malloc-based tool is blind to it, which is why this took so long.

Why the usual tools find nothing

.sys.mem reports anon-committed: 4.3 GB with sys-mapped: 0 — Rayforce
commits its own pools via mmap rather than through the malloc family. So
heaptrack, valgrind memcheck and LeakSanitizer all see essentially nothing:
heaptrack recorded 32 allocations total from a process that had grown
400 MB. Anyone else chasing a Rayforce memory question should skip straight to
valgrind --tool=massif --pages-as-heap=yes, which tracks the page-level
syscalls and does resolve it.

The stack

Peak snapshot, 317 MB total, of which 74.06% (234,881,024 B) is one path:

100.00% (317,161,472B) (page allocation syscalls) mmap/mremap/brk
->95.63% __mmap64 (mmap64.c:58)
  ->95.22% ray_vm_alloc_aligned (src/core/platform.c:169)
    ->95.22% heap_add_pool (src/mem/heap.c:575)
      ->95.22% ray_alloc (src/mem/heap.c:1494)
        ->74.06% (234,881,024B) ray_scratch_realloc (src/mem/heap.c:1951)
          ->74.06% ray_vec_append_raw (src/vec/vec.c:334)
            ->74.06% ray_insert (src/ops/query.c:15176)
              ->74.06% ray_eval -> vm_exec -> call_fn1
                ->74.06% ray_timers_fire_expired (src/core/timer.c:191)
                  ->74.06% ray_poll_run_for (src/core/epoll.c:351)

So: insert into a global, reached from a timer callback, growing a
scratch buffer through ray_scratch_realloc, each growth taking a fresh pool
from heap_add_pool — and the pools are evidently not given back.

(The other two branches at 10.58% each are one-offs — register_binary_op at
ray_lang_init, and a group-by — flat, not the leak.)

Reproducer

A ~30-line publisher plus our real service. The publisher .mc.pubs 900-row
batches every 37 ms with a stride that guarantees a sequence gap on every
message, so the consumer records findings continuously:

 69 MB  findings=100000  depth=54000
124 MB  findings=100000  depth=53100
179 MB  findings=100000  depth=54000
234 MB  findings=100000  depth=54000
290 MB  findings=100000  depth=54900
345 MB  findings=100000  depth=54000
400 MB  findings=100000  depth=54900

Linear at ~73 MB/min with both tables flat — 7x our production rate,
because the synthetic feed faults on every message. Happy to send the
publisher and consumer scripts if useful.

What I could not reduce it to

Neither of these leaks, over thousands of firings:

• insert of a fixed 1000-row batch into a global from a timer callback
• the same with the batch size varying per fire (500 … 80,000 rows)

So it is not simply insert-from-a-timer, nor scratch regrowth from varying
sizes alone. Given the stack names the function and line, I stopped there
rather than keep guessing at the triggering condition — heap.c:1951 with a
view of when a scratch pool is retired will be far quicker from your side than
from mine.

[singaraiona]

Thanks for the unusually careful report — the elimination work in the three
comments above saved most of the search. Here is what I can add from the
inside, including two things that are ruled out and one gap in the evidence
that you can close yourself in a minute.

I could not reproduce it

Three reproducers, all flat. Each ran ~2,500 messages / 950 passes:

Setup	Result
Local batches: 37 ms ingest timer + 500 ms pass (group-by, anti-join, inner-join, insert, trim)	bytes-allocated 3.84 MB flat, anon-committed 32 MB flat
Real .mc.sub delivery from a separate tickerplant process, same pass	flat
Same, with findings driven to its 100k cap and trimmed on every pass	8.03 MB dead flat

The third is the configuration your massif stack points at — insert into a
capped global, from a timer, over IPC-delivered batches — and it does not
grow. A 200-iteration parallel group-by loop over a 2M-row table at -c 8
was also flat, so worker heaps and the cross-thread free path are not it
either.

Ruled out: the symbol intern table

This was my best hypothesis, because it fits the shape exactly: the global
intern table is append-only and immortal, .sys.gc never touches it, and
.mem.objsize cannot see it — the retained-size walker returns early for
SYM vectors, since the domain is not a value. Unbounded growth against flat
reachable data is its signature.

Your own numbers rule it out. The intern table is not pool-backed, and
sys-current counts heap pools plus every other mapping, so

sys-current   4,330,172,416
anon-committed 4,328,521,728
              --------------
difference        1,650,688   <- every non-pool mapping, intern table included

1.65 MB. Not it. And since anon-committed ~= bytes-allocated + slack, the
3.76 GB is genuinely live buddy blocks, not fragmentation and not a
deferred-accounting artifact.

The massif stack is weaker evidence than it reads as

--pages-as-heap at the peak snapshot attributes pages currently mapped, by
the stack that mapped them. Standard pools are never unmapped — .sys.gc
releases their physical pages with madvise but the mapping stays — so your
234 MB (exactly 7 x 32 MB pools) says that insert-from-a-timer is the site
that triggered the pool adds, i.e. the largest single allocator on that
path. It does not establish that the retained objects were allocated there.
Worth knowing before you narrow further on heap.c:1951.

Your note about which tools are blind here is right and worth keeping: the
allocator commits its own pools directly, so malloc-based tooling sees almost
nothing.

The gap: you measured 3 of 407 globals

findings, depth and last_msg sum to 11.4 MB, and .sys.env reports 407
globals. The other 404 are unmeasured. A closure capturing a table, an
accumulating cache or a retained history would sit there and account for the
whole gap.

(env 0) returns the entire global environment as a dict, so you can dump
per-global retained sizes today:

(set d (env 0))
(set ks (key d))
(set szs (as 'I64 (map (fn [k] (.mem.objsize (at d k))) ks)))
(set m (.sys.mem))
(println "globals=% sum-objsize=% bytes-allocated=% gap=%"
  (count ks) (sum szs) (at m 'bytes-allocated) (- (at m 'bytes-allocated) (sum szs)))
(show (take (xdesc (table [name bytes] (list ks szs)) 'bytes) 20))

Run it twice, 60 s apart, on the live process. A gap of ~1 MB is normal
(interpreter internals, lambda bodies, namespace dicts). If the gap is 3.7 GB
and no global moves between the two samples, the memory is unreachable from
the environment and it is ours; if a global grows, that names it. Either
answer halves the problem, which is more than I can do from here without your
scripts.

Two caveats on reading it: sizes are summed per global, so structure shared
between two globals is counted twice; and SYM columns do not include their
domain, per the above.

Please do send the publisher and consumer scripts — my reproducers differ from
your service in exactly the two ways you already identified (wire schema
width, and concurrent ingest while the 5 s pass runs), and I would rather test
yours than keep guessing at the shape. Happy to share mine if useful.

[vbmithr]

Ran the env dump. The memory is unreachable from the global environment —
your criterion, your answer.

All 407 globals, three samples

             sum-objsize   bytes-allocated          gap
t0            42,413,171       117,822,976    75,409,805
t+3min        44,123,171       259,154,304   215,031,133
t+4.5min      44,393,171       374,765,568   330,372,397

sum-objsize is flat: +2.0 MB across the window, and that is depth
oscillating with its buffer. bytes-allocated triples. The gap absorbs
essentially 100% of the growth.

No global moves. The top of xdesc is byte-identical between samples —
findings sits at exactly 5,700,496 in all three, and w, fresh, faults,
msgs, g do not change a byte:

name          bytes        (identical across all three samples)
w           8,926,521
fresh       8,870,092
faults      6,925,996
msgs        5,953,948
g           5,832,400
findings    5,700,496
newest         57,021
depth          45,496 -> 765,496   (the only mover; buffer between passes)
last_msg       38,325

So: gap growing, nothing in the environment growing. Per your reading, it is
unreachable and it is yours. Noted also that your sys-current minus
anon-committed arithmetic rules out the intern table — agreed, and thank you
for the caution on the massif stack; I will not read it as more than "this is
the path that triggered the pool adds".

One thing the dump taught me about my own service

Those large intermediates — w, fresh, faults, msgs, g, ~36 MB
together — are pass-locals that are globals because Rayfall's set is global.
They are overwritten every pass and they do not grow, so they are not the
leak, but they may be the difference from your reproducers: our pass leaves
several multi-MB tables bound at global scope, re-bound to a different size
each pass.

I tried exactly that as a minimal case — a timer re-binding five globals to
tables cycling through sizes 1,000 … 150,000 rows, 1,700 firings — and it is
flat. So that alone is not it either. Adding it to the ruled-out list rather
than offering it as a theory.

The publisher

~20 lines, no dependencies. Every message carries a sequence gap, so the
consumer records findings continuously — that is what makes it leak fast
(73 MB/min, 7x our production rate, where the real feed faults rarely):

(set syms (map (fn [i] (parse (format "INST%" i))) (til 773)))
(set depth (table [ts_recv instrument first_id last_id action f_last f_snapshot]
             (list (take [2000.01.01D00:00:00.000000000] 0) (take ['x] 0)
                   (take [0] 0) (take [0] 0) (take ['x] 0)
                   (take [true] 0) (take [true] 0))))
(set n 0)
(set mk
  (fn [k]
    (do (set ids (* 3 (+ k (til 900))))   ;; stride 3 => a gap on every message
        (table [ts_recv instrument first_id last_id action f_last f_snapshot]
          (list (take (enlist (timestamp 0)) 900) (take syms 900)
                ids ids (take ['U] 900)
                (take [true] 900) (take [false] 900))))))
(.time.timer.set 37 0 (fn [_t] (do (.mc.pub 'depth (mk (* n 900))) (set n (+ n 1)))))

Run it with -p 5082, then start the consumer pointed at that port.

The consumer

It is our production checker, ~420 lines, and I would rather not paste an
internal file into a public issue without asking its owner first — I am doing
that now and expect it is fine. Its shape, in case you want to build it
before I get an answer:

• upd on .mc.sub "depth" projects 7 columns out of the delivered batch and
  inserts into a global depth, stamping a monotonic rid
• a 5 s timer runs the pass: select by [instrument last_id] with four
  aggregates, left-join against select by instrument, several select where over the result, then insert into a findings global, trimmed to
  100k with a negative take
• findings rows are built column-wise from the group-by output; faults are
  formatted with format and printed, capped per pass
• crucially the pass runs while ingest continues — 27 messages/s arriving
  against a 5 s pass over a 50k-row buffer

If you would rather have mine than rebuild it, say so and I will get
permission and attach it.

[vbmithr]

Permission granted — here is the consumer, verbatim. With the publisher from
my previous comment it reproduces at ~73 MB/min.

To run:

rayforce -c 2 -p 5082 ck_pub.rfl          # publisher from the previous comment
RFTP_TP_PORT=5082 rayforce -c 2 -p 5084 checker.rfl

Then watch it from a third process:

(set h (.ipc.open "localhost:5084" 10000))
(println (div (at (.ipc.send h "(.sys.mem)") 'bytes-allocated) 1048576) " MB"
         "  findings=" (.ipc.send h "(count findings)")
         "  depth="    (.ipc.send h "(count depth)"))

findings pins at its 100k cap within a minute and depth oscillates around
50k; bytes-allocated then climbs linearly and does not stop. Two knobs if
you want to move the rate: RFTP_CHECK_MS (pass interval, default 5000) and
the (* 3 ...) stride in the publisher's mk — stride 1 produces no gaps and
therefore almost no findings, which is the closest thing I have to an
off switch for the leak.

The parts your reproducers did not have, as far as I can tell: the seven-column
wire schema, and ingest continuing throughout a 5 s pass that group-bys and
joins over a ~50k-row buffer before inserting into a second, capped global.

;; ---------------------------------------------------------------------------
;; Depth completeness checker -- an incremental subscriber off the tickerplant.
;;
;; Start with:
;;   rayforce -p 5012 src/rftp/rfl/checker.rfl < /dev/null
;;
;; The question this answers is not "is the book right" but the earlier one:
;; *did we receive everything the venue sent*. A book that is wrong because a
;; message never arrived looks exactly like a book that is wrong because the
;; fold is buggy, and only one of those is worth waking someone for -- so
;; completeness is checked first, on its own, against the venue's own sequence
;; numbers.
;;
;; Why here and not in the FH worker: the worker already validates sequences
;; live (bin_binance/conn_data.ml, `up_sequence`), but all it can do with a
;; break is bump a counter. It cannot say how many ids were lost, whether the
;; break lines up with a reconnect, or whether a message arrived more than
;; once. Those are shifts, group-bys and joins over stored history, which is
;; what this process is for.
;;
;; State, not a window: every row is checked exactly once. `depth` buffers the
;; rows received since the last pass, and `last_msg` carries each instrument's
;; most recent message from one pass to the next, which is all a continuity
;; check needs of the past. A pass costs what arrived during the interval, not
;; a window of history -- and an instrument quiet for an hour is checked
;; against its last message as surely as a busy one. Both are in-memory state
;; and are lost when the checker restarts.
;; ---------------------------------------------------------------------------

(set TP_HOST (if (== (.os.getenv "RFTP_TP_HOST") "") "127.0.0.1" (.os.getenv "RFTP_TP_HOST")))
(set TP_PORT (if (== (.os.getenv "RFTP_TP_PORT") "") "5010" (.os.getenv "RFTP_TP_PORT")))

(set CHECK_MS (if (== (.os.getenv "RFTP_CHECK_MS") "")
                  5000
                  (as 'I64 (.os.getenv "RFTP_CHECK_MS"))))
;; Findings kept, newest first. A feed that is broken everywhere must not grow
;; this process without bound; past the cap the oldest findings are dropped.
(set MAX_FINDINGS (if (== (.os.getenv "RFTP_CHECK_MAX_FINDINGS") "")
                      100000
                      (as 'I64 (.os.getenv "RFTP_CHECK_MAX_FINDINGS"))))

;; --- Received stream ------------------------------------------------------
;; A projection of the TP's `depth`, not a copy of it: gap detection needs the
;; venue id range, the message boundary and the reset markers, and nothing
;; else. `rid` is assigned on receipt and is the *arrival* order -- the check
;; must never order by the venue ids it is verifying, or a duplicated id would
;; sort itself into looking correct.
;; For BIN, `first_id` is Binance U. For BFU/BFC, where continuity is
;; current pu = previous u, the publisher stores pu + 1 as `first_id`; this
;; lets the same range comparison implement both venue rules.
;;
;; Between passes this holds the rows not checked yet: everything received
;; since the last pass, plus the rows of any message still arriving then.
(set depth
  (table [rid ts_recv instrument first_id last_id action f_last f_snapshot]
    (list (take [0] 0)
          (take [2000.01.01D00:00:00.000000000] 0)
          (take ['x] 0)
          (take [0] 0)
          (take [0] 0)
          (take ['x] 0)
          (take [true] 0)
          (take [true] 0))))

(set next_rid 0)

;; Each instrument's most recent message, in the shape `messages` produces.
;; It is the predecessor the first message of the next pass is compared with.
(set last_msg
  (table [instrument first_id last_id action f_snapshot rid ts_recv]
    (list (take ['x] 0)
          (take [0] 0)
          (take [0] 0)
          (take ['x] 0)
          (take [true] 0)
          (take [0] 0)
          (take [2000.01.01D00:00:00.000000000] 0))))

;; The highest rid a completed pass has consumed. Every message of a pass has
;; a larger rid than every row of `last_msg`, which is how the two are told
;; apart once they have been ordered together.
(set checked (- 0 1))

;; --- Connection events ----------------------------------------------------
;; What a discontinuity has to be explained by. A reconnect legitimately breaks
;; the id sequence; a lost message does not. Until the FH publishes these, this
;; table stays empty and every reset is reported unexplained -- which is the
;; honest reading, not a false alarm to suppress.
(set conn_events
  (table [ts instrument event]
    (list (take [2000.01.01D00:00:00.000000000] 0)
          (take ['x] 0)
          (take ['x] 0))))

;; --- Findings -------------------------------------------------------------
;; One row per anomaly. `kind` is:
;;   'gap        ids between two messages that never arrived  (missing = how many)
;;   'overlap    a message whose range starts at or before the previous end
;;   'reset      a book reset / snapshot boundary -- expected, but accounted
;;   'multi_last a message terminated more than once, or delivered twice
;;   'no_last    a message that never terminated
;; `explained` remains false until conn_events has a producer and a specified
;; time-correlation rule.
(set findings
  (table [ts instrument kind prev_last cur_first missing rid explained]
    (list (take [2000.01.01D00:00:00.000000000] 0)
          (take ['x] 0)
          (take ['x] 0)
          (take [0] 0)
          (take [0] 0)
          (take [0] 0)
          (take [0] 0)
          (take [true] 0))))

;; --- Ingest ---------------------------------------------------------------
;; The batch arrives in the TP's wire schema; project it and stamp arrival
;; order. `update` cannot introduce a new column, so the projection is written
;; out rather than derived.
(set project
  (fn [b]
    (table [rid ts_recv instrument first_id last_id action f_last f_snapshot]
      (list (+ next_rid (til (count b)))
            (get b 'ts_recv)
            (get b 'instrument)
            (get b 'first_id)
            (get b 'last_id)
            (get b 'action)
            (get b 'f_last)
            (get b 'f_snapshot)))))

(set upd
  (fn [topic _seq batch]
    (if (== topic "depth")
        (do (insert 'depth (project batch))
            (set next_rid (+ next_rid (count batch))))
        null)))

;; --- Message integrity ----------------------------------------------------
;; Every message must terminate exactly once. Counting f_last rows per
;; (instrument, last_id) catches a truncated message -- a file rotation or an
;; Aeron term boundary cutting one in half, which no sequence check would see
;; because the ids either side still line up -- a duplicated boundary, and a
;; message delivered twice within one pass.
;;
;; A group without f_last is not yet a fault: the bridge batches by row count,
;; so a pass can land between two rows of one message. Only once a later row
;; for the same instrument has arrived (`top` below `inst_top`) is it really
;; cut. Until then its rows stay in `depth` for the next pass.
(set message_groups
  (fn []
    (left-join [instrument]
      (select {from: depth
               by: [instrument last_id]
               lasts: (sum f_last)
               top: (max rid)
               ts_recv: (min ts_recv)})
      (select {from: depth by: instrument inst_top: (max rid)}))))

(set find_multi_last
  (fn [g] (select {from: g where: (> lasts 1)})))

(set find_no_last
  (fn [g] (select {from: g where: (and (== lasts 0) (< top inst_top))})))

;; Every group but the messages still arriving.
(set find_consumed
  (fn [g] (select {from: g where: (not (and (== lasts 0) (== top inst_top)))})))

;; --- Continuity -----------------------------------------------------------
;; Message-level view of the buffer: one row per venue message, taken from its
;; terminating f_last row. Identical (instrument, [U,u]) copies collapse here,
;; so a range delivered twice counts once for continuity; integrity names it.
;; A message that never terminated has no row here, so its ids read as missing
;; -- which, for a feed cutting messages in half, they are.
(set messages
  (fn []
    (select {from: (select {from: depth where: (== f_last true)})
             by: [instrument first_id last_id action f_snapshot]
             rid: (min rid)
             ts_recv: (min ts_recv)})))

;; `window` with a two-row trailing frame over this pass's messages and each
;; instrument's `last_msg` gives every message the previous message for the
;; same instrument: `prev`/`prev_first` are its range, and `n` is 1 only for
;; an instrument never seen before. Rows with rid at or below `checked` are the
;; carried predecessors, not messages of this pass.
(set continuity
  (fn [msgs]
    (window {from: (concat last_msg msgs)
             part: [instrument]
             order: [rid]
             frame: 2
             funcs: {prev: (first last_id) prev_first: (first first_id) n: (count last_id)}})))

;; The newest message of every instrument in [w], carried predecessors
;; included, so an instrument silent this pass keeps its last message.
(set latest
  (fn [w]
    (do (set newest (inner-join [instrument rid] w
                                (select {from: w by: instrument rid: (max rid)})))
        (table [instrument first_id last_id action f_snapshot rid ts_recv]
          (list (get newest 'instrument)
                (get newest 'first_id)
                (get newest 'last_id)
                (get newest 'action)
                (get newest 'f_snapshot)
                (get newest 'rid)
                (get newest 'ts_recv))))))

;; A reset is a legitimate break: the venue restarted the book, either because
;; we resubscribed or because it said so. It is classified as 'reset and never
;; as 'gap, whatever the ids do across it. The bridge writes a DBN action under
;; its constructor name, so a clear is 'Clear, not DBN's 'R.
(set find_resets
  (fn [fresh]
    (select {from: fresh where: (or (== action 'Clear) (== f_snapshot true))})))

;; The same range as its predecessor: a copy of a message whose original was
;; consumed by an earlier pass. Within a pass copies collapse into one message
;; instead, and integrity sees the doubled f_last.
(set find_repeats
  (fn [fresh]
    (select {from: fresh
             where: (and (== n 2) (and (== first_id prev_first) (== last_id prev)))})))

;; Messages whose ids must follow on from their predecessor's.
(set sequenced
  (fn [fresh]
    (select {from: fresh
             where: (and (== n 2)
                         (not (or (or (== action 'Clear) (== f_snapshot true))
                                  (and (== first_id prev_first) (== last_id prev)))))})))

(set find_gaps
  (fn [fresh] (select {from: (sequenced fresh) where: (> first_id (+ prev 1))})))

(set find_overlaps
  (fn [fresh] (select {from: (sequenced fresh) where: (<= first_id prev)})))

;; --- Recording ------------------------------------------------------------
(set finding_rows
  (fn [kind ts_col inst_col prev_col cur_col missing_col rid_col]
    (table [ts instrument kind prev_last cur_first missing rid explained]
      (list ts_col
            inst_col
            (take (enlist kind) (count inst_col))
            prev_col
            cur_col
            missing_col
            rid_col
            (take [false] (count inst_col))))))

(set record
  (fn [rows]
    (if (> (count rows) 0) (insert 'findings rows) null)))

(set record_sequence
  (fn [t kind]
    (do (set zeros (take [0] (count t)))
        (record (finding_rows kind
                              (get t 'ts_recv)
                              (get t 'instrument)
                              (get t 'prev)
                              (get t 'first_id)
                              (if (== kind 'gap)
                                  (- (get t 'first_id) (+ (get t 'prev) 1))
                                  zeros)
                              (get t 'rid))))))

;; Integrity findings are properties of a message, not of an arrival position:
;; recorded with the message's last_id in `cur_first` and rid 0, because no
;; single row owns the fault.
(set record_integrity
  (fn [t kind]
    (do (set zeros (take [0] (count t)))
        (record (finding_rows kind
                              (get t 'ts_recv)
                              (get t 'instrument)
                              zeros
                              (get t 'last_id)
                              zeros
                              zeros)))))

(set trim_findings
  (fn []
    (if (> (count findings) MAX_FINDINGS)
        (set findings (select {from: findings take: (- 0 MAX_FINDINGS)}))
        null)))

;; --- One pass -------------------------------------------------------------
;; [g] is the pass's message groups: a repeat whose boundary integrity has
;; already counted twice -- a snapshot and a delta of the same range, which
;; differ in action and so do not collapse -- is one fault, reported once.
(set check_continuity
  (fn [msgs g]
    (do (set w (continuity msgs))
        (set fresh (select {from: w where: (> rid checked)}))
        (record_sequence (find_gaps fresh) 'gap)
        (record_sequence (find_overlaps fresh) 'overlap)
        (record_sequence (find_resets fresh) 'reset)
        (record_integrity (anti-join [instrument last_id] (find_repeats fresh) (find_multi_last g))
                          'multi_last)
        (set last_msg (latest w)))))

;; --- Logging --------------------------------------------------------------
;; A fault is logged as it is found, one line each, saying what happened in
;; the terms of the feed: which instrument, which ids. Resets are not faults
;; -- every book snapshot is one -- so they are only counted. A heartbeat
;; every REPORT_MS says the checker is alive and what it
;; has seen, so a quiet log reads as a healthy feed rather than a stuck
;; checker.
(set REPORT_MS (if (== (.os.getenv "RFTP_CHECK_REPORT_MS") "")
                   60000
                   (as 'I64 (.os.getenv "RFTP_CHECK_REPORT_MS"))))
(set PASSES_PER_REPORT (max 1 (div REPORT_MS CHECK_MS)))

;; A pass that finds more than this describes the first ones and counts the
;; rest; `(summary)` and `findings` hold them all.
(set MAX_FAULT_LINES 10)

(set passes 0)
(set seen_rows 0)
(set seen_msgs 0)
(set seen_faults 0)
(set seen_resets 0)
(set total_faults 0)
(set total_missing 0)

(set describe
  (fn [f i]
    (do (set f_kind (at (get f 'kind) i))
        (set f_inst (at (get f 'instrument) i))
        (set f_prev (at (get f 'prev_last) i))
        (set f_cur (at (get f 'cur_first) i))
        (set f_ts (at (get f 'ts) i))
        (if (== f_kind 'gap)
            (format "checker: gap % -- previous message ended at id %, next started at %: % id(s) %..% never arrived; received %"
                    f_inst f_prev f_cur (at (get f 'missing) i) (+ f_prev 1) (- f_cur 1) f_ts)
            (if (== f_kind 'overlap)
                (format "checker: overlap % -- message starting at id % repeats ids up to %, already received; received %"
                        f_inst f_cur f_prev f_ts)
                (if (== f_kind 'multi_last)
                    (format "checker: duplicate % -- message ending at id % was received or terminated more than once; received %"
                            f_inst f_cur f_ts)
                    (format "checker: truncated % -- message ending at id % never got its last row; received %"
                            f_inst f_cur f_ts)))))))

(set log_faults
  (fn [faults]
    (do (set shown (min (count faults) MAX_FAULT_LINES))
        (map (fn [i] (println (describe faults i))) (til shown))
        (if (> (count faults) shown)
            (println (format "checker: ... and % more faults this pass; (summary) lists them all"
                             (- (count faults) shown)))
            null))))

(set heartbeat
  (fn []
    (do (println (format "checker: last %s -- % rows, % messages, % instruments; % faults, % resets | since start: % faults, % ids missing"
                         (div (* passes CHECK_MS) 1000)
                         seen_rows seen_msgs (count last_msg)
                         seen_faults seen_resets
                         total_faults total_missing))
        (set passes 0)
        (set seen_rows 0)
        (set seen_msgs 0)
        (set seen_faults 0)
        (set seen_resets 0))))

;; [added] is what this pass recorded, [rows] and [nmsgs] what it consumed.
(set report_pass
  (fn [added rows nmsgs]
    (do (set faults (select {from: added where: (!= kind 'reset)}))
        (log_faults faults)
        (set total_faults (+ total_faults (count faults)))
        (set total_missing
             (+ total_missing (sum (get (select {from: faults where: (== kind 'gap)}) 'missing))))
        (set passes (+ passes 1))
        (set seen_rows (+ seen_rows rows))
        (set seen_msgs (+ seen_msgs nmsgs))
        (set seen_faults (+ seen_faults (count faults)))
        (set seen_resets (+ seen_resets (- (count added) (count faults))))
        (if (>= passes PASSES_PER_REPORT) (heartbeat) null))))

;; --- One pass -------------------------------------------------------------
;; Consumes the buffer: every message in it is checked once, and only the rows
;; of messages still arriving are kept for the next pass.
(set run_checks
  (fn []
    (do
      (set before (count findings))
      (set rows (- next_rid (+ checked 1)))
      (set g (message_groups))
      (record_integrity (find_multi_last g) 'multi_last)
      (record_integrity (find_no_last g) 'no_last)
      (set msgs (messages))
      (if (> (count msgs) 0) (check_continuity msgs g) null)
      (set depth (anti-join [instrument last_id] depth (find_consumed g)))
      (set checked (- next_rid 1))
      (report_pass (select {from: findings take: (- 0 (- (count findings) before))})
                   rows
                   (count msgs))
      (trim_findings)
      (count findings))))

;; --- Querying -------------------------------------------------------------
(set summary
  (fn [] (select {from: findings by: [instrument kind] n: (count kind)})))

(set unexplained
  (fn [] (select {from: findings where: (== explained false)})))

(set missing_total
  (fn [] (sum (get (select {from: findings where: (== kind 'gap)}) 'missing))))

;; --- Subscribe and run ----------------------------------------------------
(set tp (.ipc.open (format "%:%" TP_HOST TP_PORT)))
(.ipc.send tp (list '.mc.sub "depth" null))
(println (format "checker: subscribed to depth at %:%" TP_HOST TP_PORT))

(.time.timer.set CHECK_MS 0 (fn [t] (run_checks)))

(println (format "checker: ready -- a pass every % ms, a heartbeat every % ms" CHECK_MS REPORT_MS))

[vbmithr]

Minimal reproducer, 24 lines. No IPC, no multicast, no findings table. And it explains why every measurement in this thread said "nothing is growing".

The shape

State carried across passes is projected out of a table derived from that same state. Each pass: last_msg → concat → window → inner-join → project columns → last_msg.

(set syms (map (fn [i] (parse (format "INST%" i))) (til 773)))
(set last_msg (table [instrument rid first_id last_id]
                (list (take ['x] 0) (take [0] 0) (take [0] 0) (take [0] 0))))
(set n 0)
(set mknew
  (fn [k] (table [instrument rid first_id last_id]
            (list syms (+ k (til 773)) (+ k (til 773)) (+ k (til 773))))))
(.time.timer.set 50 0
  (fn [_t]
    (do (set w (window {from: (concat last_msg (mknew (* n 773)))
                        part: [instrument] order: [rid] frame: 2
                        funcs: {prev: (first last_id) c: (count last_id)}}))
        (set newest (inner-join [instrument rid] w
                                (select {from: w by: instrument rid: (max rid)})))
        (set last_msg (table [instrument rid first_id last_id]
                        (list (get newest 'instrument) (get newest 'rid)
                              (get newest 'first_id) (get newest 'last_id))))
        (set n (+ n 1))
        (if (== (- n (* 200 (div n 200))) 0)
            (println "passes=" n " last_msg=" (count last_msg)
                     " objsize=" (.mem.objsize last_msg))
            null))))

Result

rss= 55MB  passes= 600  last_msg=773  objsize=25040
rss= 93MB  passes=1200  last_msg=773  objsize=25040
rss=127MB  passes=1800  last_msg=773  objsize=25040
rss=165MB  passes=2400  last_msg=773  objsize=25040

773 rows and 25,040 bytes, constant to the byte, while RSS grows ~75 MB/min.

Why everything looked clean

get/projection appears to share column storage with its parent. So each pass's last_msg transitively pins newest → w → (concat last_msg_prev …) → the previous last_msg, and so on: an unbounded chain of retained ancestors, one per pass.

That accounts for every confusing measurement here:

• .mem.objsize last_msg reports 25,040 forever, because it walks the table's own storage and not the parents it pins. My whole-environment dump was therefore flat while the gap grew — the memory is reachable, just not through anything objsize counts.
• Every primitive in isolation is clean, because none of them carries its own output back into its next input.
• My earlier reproducers all rebuilt state from fresh tables each iteration, which breaks the chain — which is exactly why they were flat.

Questions

1. Is storage sharing through get/projection intended? I assume yes, and that it is the right default.
2. If so, is there a way to force materialisation — to detach a table from its ancestors? That is the fix on our side and I could not find one. A copy-like primitive, or a documented idiom, would close this class of bug for anyone carrying state forward across iterations.
3. Would it be reasonable for .mem.objsize to report retained-including-parents, or at least expose the distinction? As it stands the accounting actively hides this.

Bisect that got here, for the record: the leak is in check_continuity. window alone is flat over 2,400 calls; message_groups + drain is flat; messages alone is flat. It is specifically the feedback of projected state into the next iteration.

[ser-vasilich]

Thanks — the 24-line reproducer is exactly the shape, and it settles the cause.

It is not an ancestor chain through projections. concat copies its inputs' columns into new vectors, and a projection through get only shares refcounted column vectors with its parent (that part is intended and is not a leak). What leaks is the input table of the window itself: the executor treated the table a window (and a sorted or limited select) is rooted on as a borrowed reference and never released it, once per call. Your reproducer grows by exactly that table — the concat result — on every pass, with last_msg staying at 773 rows and 25 040 bytes, which is why every per-global measurement was flat.

Your script, 1400 passes, bytes-allocated:

build	200 passes	1400 passes
current dev	13.3 MB	92.2 MB (+66 KB per pass)
dev + #604	288 KB	289 KB

#604 fixes it. No materialise step is needed on your side, and .mem.objsize was reporting correctly: the retained memory was not reachable from last_msg at all, it was an unreleased reference held by the executor.