diff --git a/CMakeLists.txt b/CMakeLists.txt
index 57c83a8ef..79801b719 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -47,11 +47,16 @@ ENDIF()
 # came out differently from the same source and data. Off, every x86-64 level computes the same bits
 # (measured: identical p.mtz on nine battery sets, no measurable time cost), and aarch64, where FMA is
 # always there and GCC contracts by default, follows the same arithmetic. MSVC does not contract under
-# its default /fp:precise. Host code of .cu sources gets it through nvcc; device code is not affected.
+# its default /fp:precise. Host code of .cu sources gets it through nvcc.
 IF (NOT MSVC)
     ADD_COMPILE_OPTIONS($<$<COMPILE_LANGUAGE:C,CXX>:-ffp-contract=off>)
     ADD_COMPILE_OPTIONS($<$<COMPILE_LANGUAGE:CUDA>:-Xcompiler=-ffp-contract=off>)
 ENDIF()
+# The same for device code. nvcc contracts by default and does it separately for every architecture
+# it compiles for, so a T4 (sm_75) and an RTX 5080 (sm_120) ran the same kernels to different bits
+# and a run that hit a knife edge came out differently on the two machines. Off, the architectures
+# compute the same bits (measured: identical p.hkl from sm_75 and sm_120 builds, no measurable time cost).
+ADD_COMPILE_OPTIONS($<$<COMPILE_LANGUAGE:CUDA>:--fmad=false>)
 
 SET(JFJOCH_WRITER_ONLY           OFF CACHE BOOL "Compile HDF5 writer only")
 SET(JFJOCH_INSTALL_DRIVER_SOURCE OFF CACHE BOOL "Install kernel driver source (ignored if building writer only; necessary for RPM building)")
diff --git a/docs/JFJOCH_WRITER.md b/docs/JFJOCH_WRITER.md
index 3bd9f8823..1ebca2e6e 100644
--- a/docs/JFJOCH_WRITER.md
+++ b/docs/JFJOCH_WRITER.md
@@ -21,7 +21,8 @@ Options:
 -d<path>  | --root_dir=<path>          Root directory for file writing (-R is a deprecated alias)
 -r<int>   | --zmq_repub_port=<int>     ZeroMQ port for PUSH socket to republish images
 -f<int>   | --zmq_file_port=<int>      ZeroMQ port for PUB socket for notifications on finalized files
--w<int>   | --rcv_watermark=<int>      Receiving ZeroMQ socket watermark (default = 100)
+-w<int>   | --rcv_watermark=<int>      Receiving ZeroMQ socket watermark in messages (default = 10)
+-m<int>   | --rcv_ring_mb=<int>        Receive ring buffer in MB; holds images until written and republished (default = 2048)
 -W<int>   | --repub_watermark=<int>    Republish ZeroMQ socket watermark (default = 1000)
 -b<int>   | --rcv_buffer=<int>         Receiving TCP socket buffer in bytes (only with -T; disables autotuning, capped by net.core.rmem_max)
 -v        | --verbose                  Verbose output
@@ -32,6 +33,33 @@ for example:
 jfjoch_writer -d /data tcp://dcu-address:5400 
 ```
 
+## Memory
+Every message the writer receives is placed in a ring buffer of a fixed size (`-m`, default 2048 MB,
+per writer process - `-j 8` is eight rings) and stays there until it has been written to HDF5 and, if
+republish is on, republished. The ring is the writer's memory budget for buffered images, and the
+peak use is printed at the end of every run. Size it as the number of images you can afford to have
+in flight times their compressed size: at 2048 MB that is some 250-500 JUNGFRAU 4M images but only
+50-100 JUNGFRAU 9M images, so a 9M site will want more.
+
+When the ring is full the writer stops reading from the socket and the sender has to wait. How long
+it is prepared to wait depends on the stream:
+
+- **ZeroMQ.** The socket queue fills up to its watermark (`-w`, a count of messages, kept small for
+  that reason) and the broker's send blocks for one second. After that the broker drops the image,
+  logs the fact, and sends without waiting for the rest of the run, so every later image that finds
+  the queue full is dropped too. A writer that stalls for longer than its ring covers therefore
+  loses images on ZeroMQ; the ring has to be large enough for the stalls of the filesystem it writes
+  to. (Before the ring the same stall was absorbed by queues of about 1500 messages, whatever their
+  size - a budget nobody had chosen.)
+- **TCP/IP.** The broker sees the writer's BUSY heartbeats and waits, giving up only after a minute
+  in which not a single image was written.
+
+Republish holds its own images in the same ring. Because the ring is reclaimed in arrival order, an
+image kept for a slow republish consumer would hold back everything received after it, so republish
+may hold at most a quarter of the ring; beyond that images are simply not republished, exactly as
+when its queue is full. A single message larger than the whole ring - a START or END message of a
+very long run - is held outside it.
+
 ## Status and cancellation
 When a data collection is finalized, each writer reports its outcome back to `jfjoch_broker` over
 the writer notification socket — a ZeroMQ address the broker passes in the START message
diff --git a/docs/RUGNUX.md b/docs/RUGNUX.md
index 6a38593f8..5b4b945cd 100644
--- a/docs/RUGNUX.md
+++ b/docs/RUGNUX.md
@@ -192,6 +192,8 @@ One page per job, so the answer needed is near the top of a short page:
   anisotropy section.
 - [Advanced usage](RUGNUX_ADVANCED.md) — reference data and the indexing ambiguity, model
   validation, re-merging, and the full command-line option tables.
+- [Models and reference data](RUGNUX_MODEL_AND_REFERENCE.md) — every model and reference option,
+  and decision by decision what a model can and cannot change.
 - [Detector calibration](RUGNUX_CALIBRATION.md) — the geometry from a calibrant's powder rings
   (`--mode calibration`).
 - [CPU/GPU data analysis](CPU_DATA_ANALYSIS.md) — the algorithms behind all of it.
diff --git a/docs/RUGNUX_ADVANCED.md b/docs/RUGNUX_ADVANCED.md
index 719b903dd..4bcf8edd9 100644
--- a/docs/RUGNUX_ADVANCED.md
+++ b/docs/RUGNUX_ADVANCED.md
@@ -147,6 +147,9 @@ determined on, CCP4 style (`k,l,h` is h'=k, k'=l, l'=h); `h,k,l` where nothing m
 
 ## Validating against a model (`rugnux --model`)
 
+Which decisions a model can influence, and which it cannot, is set out decision by decision in
+[Models and reference data](RUGNUX_MODEL_AND_REFERENCE.md).
+
 Given an atomic model of the same structure, `--model model.pdb` scales the model structure
 factors to the merged amplitudes — fitting a flat bulk-solvent contribution and an overall
 anisotropic *B* — and reports **R-work / R-free** and the mean 2mFo−DFc density at the atom centres.
diff --git a/docs/RUGNUX_MODEL_AND_REFERENCE.md b/docs/RUGNUX_MODEL_AND_REFERENCE.md
new file mode 100644
index 000000000..1cc62e8c2
--- /dev/null
+++ b/docs/RUGNUX_MODEL_AND_REFERENCE.md
@@ -0,0 +1,333 @@
+# Models and reference data
+
+What Rugnux does with outside knowledge — an atomic model, a predicted model, a reference dataset, a
+cell or a space group you already know — and, decision by decision, what that knowledge is and is not
+allowed to change. The first section is the short version; the tables after it are the precise one;
+the rest walks through a run step by step.
+
+```{contents} On this page
+:local:
+:depth: 2
+```
+
+## In short
+
+- **Rugnux needs nothing but the images.** Indexing, the cell, the space group, the resolution cut,
+  scaling, merging and the error model all come from the data. Everything on this page is optional.
+- **A model (`--model`, `--mr`) does not change how your data are processed.** The images are
+  indexed, integrated, scaled and merged the same way with or without it, and the space group is
+  determined before the model is used. The merged intensities and their sigmas in
+  `<prefix>.mtz` are the same numbers either way.
+- **The model can only relabel, and only if the data accept it.** It is first tested against
+  copies of itself placed in random orientations. A model that fits no better than those changes
+  nothing at all: the reflection files are byte for byte what a run without a model writes. A model
+  that does fit may change three *labels* in `<prefix>.mtz`: the **hand** of the space group
+  (P4₁2₁2 against P4₃2₁2, which the intensities cannot tell apart), the **indexing** where the
+  lattice allows two equally valid ones, and the **axes** the files are written on. No
+  intensity is changed, no reflection is added or removed, and no other space group is ever taken.
+- **A different space group chosen by the model goes to a separate file.** If the model prefers
+  another space group, or a twinned subgroup, that merge is written *beside* the answer as
+  `<prefix>_mr.mtz`/`.cif`, with the model as placed there. `<prefix>.mtz` stays the data's own
+  answer.
+- **A reference dataset (`-z`) and a space group or cell you give (`-S`, `-C`) are different:** they
+  are taken as facts about the crystal. They fix the space group and the cell before processing,
+  so with them the space group is yours, not the data's.
+- **One exception to "processed the same":** on **serial stills** given `--model` together with
+  `-C` and `-S`, the model is used during integration to choose, crystal by crystal, between the two
+  equally valid indexings — the same job a reference dataset does there (see
+  [Serial stills](#serial-stills-the-one-place-a-model-acts-before-the-merge)).
+
+## Every option that brings in a model or reference data
+
+| Option | What you give | What Rugnux uses it for | Can it change `<prefix>.mtz`? |
+|---|---|---|---|
+| *(none)* | — | Everything from the data | — |
+| `--model <file>` | An atomic model, PDB or mmCIF, gzipped or not, recognised by content: an experimental structure or a prediction | Validation after the merge (R-work/R-free, CC(model, data), maps, the placed model); a rigid-body placement checked against a null, with molecular replacement where the rigid body does not place it, and straight away for a prediction | Labels only (hand, indexing, axes), and only where the model fits. Other space groups → `<prefix>_mr.*` |
+| `--mr <file>` | The same kinds of model | As `--model`, but molecular replacement straight away, without the rigid body. `--mode mx` only, not together with `--model` | Hand label only. Other space groups → `<prefix>_mr.*` |
+| PAE file beside a prediction | `AF-<id>-F1-predicted_aligned_error_v<n>.json` next to `AF-<id>-F1-model_v<n>.cif` (read automatically, no option) | Cuts the prediction into rigid domains for molecular replacement | No (a domain alone can only confirm the answer's group) |
+| `-z, --reference <file>` (`--reference-mtz`) | Merged data of the same crystal form: an MTZ, or a PDB structure-factor mmCIF (`-sf.cif`) | Fixes the space group (in its own setting) and the cell (softly), puts the merge on its axes, resolves the indexing ambiguity, hands over its R-free set, reports CC<sub>ref</sub> on stills. Never a scale anchor | **Yes**: the space group, the axes, the indexing and the free set are the reference's |
+| `--reference-column <label>` | An MTZ column label | Which column of the reference is read | Through `-z` only |
+| `-S, --space-group` | A space group | Fixes it: no space-group search | **Yes**: the group is yours (a fitting model may still relabel its hand) |
+| `-C, --unit-cell` | A cell | Indexing looks for this cell (within tolerance); the FFT search range is widened to cover it; on stills it turns on the shared-geometry refinement and, with `-S` and `--model`, the per-image use of the model | **Yes**: the cell, and the axis order the files are written in |
+| `--force-rotation-lattice` | Nine numbers: a lattice | Skips the first-pass indexing and uses this lattice | **Yes**: the lattice |
+| `--report-resolution <dmin>[,<dmax>]` | Another program's resolution range | A second statistics table at that range (`REFRES_*`) | No — report only |
+
+`-S`, `-C`, `-z` and `--force-rotation-lattice` are *assertions*: Rugnux takes them as true and
+does not test them against a null. Use them when you know; leave them out to let the data decide
+(a de-novo run is the recommended default — see [Running Rugnux](RUGNUX_TUTORIAL.md#rotation-data)).
+`--model` and `--mr` are *hypotheses*: Rugnux tests them, and what it lets them change is limited
+and stated below. In `--mode calibration` a cell given with `-C` names the calibrant instead
+([Detector calibration](RUGNUX_CALIBRATION.md)); nothing on this page applies there.
+
+## Four kinds of influence
+
+Every decision and every output of a run falls in one of four classes. The decision table below uses
+these letters.
+
+- **(a) Data only.** Never influenced by any model. The same with and without `--model`/`--mr`.
+- **(b) Model-informed label in `<prefix>.mtz`.** A choice between alternatives the data *cannot*
+  separate and that are all equally valid descriptions of the same measurement: the hand of an
+  enantiomorphic pair, an indexing ambiguity, the setting (axes). The model may make it only after it
+  has passed its null; the intensities do not change. Reported as `SPACE_GROUP_ENANTIOMORPH=
+  ASSUMED_FROM_MODEL`, `MODEL_DECISIONS_TAKEN=`, `MODEL_ENANTIOMORPH_ADOPTED=`,
+  `MODEL_INDEXING_OPERATOR=`, `SETTING_SOURCE= MODEL`, and the withdrawal of
+  `INDEXING_AMBIGUITY_OPERATORS=`.
+- **(c) Model-driven, separate files only.** A choice the data *could* in principle make but did not
+  make the model's way: another space group with the same rotations (other screw axes), or the twin
+  hypothesis' subgroup. Written only as `<prefix>_mr.mtz`/`.cif` with `<prefix>_mr_model.pdb`/`.cif`
+  (and named by `MODEL_CHOICE_FILE_PREFIX=`, `MODEL_CHOICE_SPACE_GROUP=`, `MR_DECISION=`). Never in
+  `<prefix>.mtz`.
+- **(d) Report-only model statistics.** Numbers about the model against the data, which nothing in
+  the pipeline acts on: `R_WORK`/`R_FREE`, `R_MODEL_SHELL_SCALED`, `CC_MODEL_*`, `MODEL_FIT*`,
+  `MODEL_PLACEMENT`, the `MR_*` keys, the anomalous sites, the maps.
+
+## Decision by decision
+
+"Reference" means `-z`. A dash means the option plays no part.
+
+| Decision | Without anything | With `-S` / `-C` | With `-z` | With `--model` / `--mr` | Class |
+|---|---|---|---|---|---|
+| Spot finding, beam centre, detector geometry | data | — | — | — | (a) |
+| Indexing, lattice, cell | data (de novo) | your cell / lattice | the reference's cell, softly | — | (a) |
+| Integration | data | — | — | — (stills: see below) | (a) |
+| Scaling, error model, ISa | data | — | — (never a scale anchor) | — | (a) |
+| Outlier rejection, frame disposition | data | — | — | — | (a) |
+| Resolution cut | data (CC1/2 fall-off) | — | — | — (`CC_MODEL_CONFIRMED_TO_D_MIN` is reported, never used) | (a) |
+| Point group, Laue class, centring | data (space-group search) | `-S` fixes it | the reference's | — | (a) |
+| Screw axes (space group within the point group) | data | `-S` | the reference's | a different choice → `<prefix>_mr.*` only | (a) in `p.mtz`, (c) beside it |
+| Hand (enantiomorph) | not decidable: `SPACE_GROUP_ENANTIOMORPH= UNDETERMINED` | `GIVEN` | `GIVEN` | relabelled to the model's where it fits, past the anomalous veto: `ASSUMED_FROM_MODEL` | (b) |
+| Indexing ambiguity (merohedral reindex) | arbitrary on rotation (`INDEXING_AMBIGUITY_OPERATORS=`); mixed on stills (warning) | — | resolved against the reference | resolved by the rigid-body model where its lead beats its null, rotation; per image on stills with `-C`/`-S`. Never by molecular replacement | (b) |
+| Setting (which axis is a, b, c) | standard setting | `-C`'s axis order; `-S`'s symbol | the reference's | the model's, where it fits and there is no `-z` (`SETTING_SOURCE= MODEL`) | (b) |
+| Twinning check (L-test, twin laws) | data | — | — | — | (a) |
+| Twin hypothesis `<prefix>_twin_<SG>.*` | data (when the group may have absorbed a twin law) | — | — | the model may also *ask* for this merge; written then only where the model chooses it | (a) / (c) |
+| Anomalous signal, Friedel handling | data (`-A`) | — | — | — (the anomalous map is report-only, and the veto on the hand) | (a) |
+| R-free test set | hash of the index | — | the reference's flags, where it matches | — (drawn again on the model's axes if the setting moved; never chosen by the model) | (a) |
+| Merged intensities and sigmas in `p.mtz` | data | — | — | — | (a) |
+| `p.mtz` space-group label | data | yours | the reference's | hand relabel only | (b) |
+| `<prefix>_mr.mtz`/`.cif`, `<prefix>_mr_model.*` | not written | — | — | another group or the twin subgroup the model chose | (c) |
+| Maps, `<prefix>_model.*`, section 5 of the report | not written | — | — | always, whatever the verdict | (d) |
+
+## A run with `--model`, step by step
+
+1. **The file is read and classified.** PDB or mmCIF, by content. A **prediction** — ModelCIF's
+   quality metrics, or no cell and one B value per residue between 0 and 100 (an AlphaFold file) —
+   is told apart from an **experimental structure** (`MR_MODEL_KIND=`). A model that cannot be read
+   does not fail the run; the report says `MODEL_VALIDATION= NOT_PERFORMED` and carries a warning.
+2. **The data are processed without the model.** Indexing, integration, scaling, the space-group
+   search, the resolution cut, the error model, outlier rejection: none of them reads the model
+   (on serial stills with `-C`/`-S`, see the next section). The only thing the model asks for here is
+   one extra merge: the merge in a subgroup (the twin hypothesis) where the indexed lattice's group
+   was promoted by a metric re-ask, or where the model's own group has fewer rotations than the
+   adopted one — so that the model can be scored there. That merge is written only if the model
+   chooses it.
+3. **Rigid-body placement** (experimental structures, not with `--mr`). The model is put into the
+   data's cell and refined as one rigid body on the working reflections; the step is kept only if
+   it lowers R-free. A model written in another description of the lattice (other axes, other
+   centring) is first put into the data's (`MODEL_CHANGE_OF_BASIS=`).
+4. **The null.** The same model is turned to nine random orientations about its own centroid and
+   placed the same way. The real fit has to beat them by 4.51 of their standard deviations on
+   R-work (3σ allowing for the null being nine draws): `MODEL_FIT= ACCEPTED`, `MODEL_FIT_SIGMA=`.
+   Otherwise `REJECTED`, and the model decides nothing. There is no fixed threshold on R
+   ([§14.5](CPU_DATA_ANALYSIS_DECISIONS.md#145-does-the-model-fit-the-null-it-is-scored-against)).
+5. **What an accepted model may do** (class b), applied to the reflections before they are written:
+   - **hand** — where the data's group is the enantiomorph of the model's, the files take the
+     model's group as their label. No reflection moves and I(+)/I(-) stay as measured. Vetoed where
+     the anomalous difference map read at the model's atoms is inverted (warning `MODEL_HAND`);
+   - **indexing** — where the lattice admits alternative indexings and no `-z` already settled it,
+     the candidate with the lowest R-free, *if* its lead over the runner-up beats the lead a random
+     placement of the same model takes (`MODEL_INDEXING_MARGIN_SIGMA=` ≥ 3). The data are
+     reindexed into the model's frame, never the model into the data's;
+   - **setting** — a model written on other axes takes the files to its setting
+     (`SETTING_SOURCE= MODEL`), so that reflections and coordinates can be used together.
+6. **The twin hypothesis.** Where a subgroup merge exists, the model is scored against it plainly
+   and as twinned by its law. The maps come from whichever merge the model fits better on R-free
+   (`MAPS_FROM=`). Where the model also fits there against its null, it has chosen the subgroup,
+   and that merge is written again as `<prefix>_mr.mtz`/`.cif` (class c).
+7. **Molecular replacement**, for a prediction, with `--mr`, or for an experimental structure the
+   rigid body did not place (in the answer's merge or in the subgroup's). As many copies as the cell
+   most probably holds are searched for in every space group the merge cannot tell apart and in the
+   twin hypothesis' subgroup, and the choice has its own null. `MR_DECISION=` is one of:
+   `AGREES` (the answer's group — nothing changes); `ENANTIOMORPH` (the answer's files take the hand
+   label, past the same anomalous veto, else `ENANTIOMORPH_VETOED`); `SPACE_GROUP` or
+   `TWIN_SUBGROUP` (written as `<prefix>_mr.*` only); `UNDECIDED` or `NOT_PLACED` (nothing is
+   written). Molecular replacement never reindexes and never sets `MODEL_FIT`. Not placing a model
+   is a result, not a warning ([Molecular replacement](RUGNUX_REPORT.md#molecular-replacement)).
+8. **Validation and files.** R-factors, CC(model, data), maps and the placed model are computed
+   for the placement that was kept, against the merge it was chosen on, and written:
+   `<prefix>_2fofc.ccp4`, `_fofc.ccp4`, `_anom.ccp4`, `_maps.mtz`, `_model.cif`/`.pdb` — or the
+   `<prefix>_mr_*` equivalents where the choice was `<prefix>_mr`. A rejected model gets its maps
+   too: that is the case where looking at it in the density is most useful.
+9. **The reflection files** — `<prefix>.mtz`, `.cif`, `.hkl`, `_unmerged.mtz` — are written last,
+   carrying only the class-(b) labels from step 5 and the hand from step 7.
+
+`MODEL_PLACEMENT=` summarises steps 3–7: `RIGID_BODY`, `MR` or `NOT_PLACED`.
+
+In `--mode scale`, `--model` is steps 3–5 only: the rigid body, its null, and the hand and indexing
+it may settle. There is no molecular replacement, no setting change and no twin hypothesis, and
+`--mr` is refused.
+
+### Serial stills: the one place a model acts before the merge
+
+On serial stills each crystal is indexed on its own, so where the lattice admits two indexings, each
+crystal lands in one of them at random and a merge averages reflections that are not equivalent.
+That has to be settled per crystal, at integration time — by the time there is a merge, the hands
+are already mixed. A reference dataset (`-z`) is the tool for it. Without one, `--model` given
+together with `-C` and `-S` supplies the reference intensities instead: structure factors computed
+from the model, against which each crystal's intensities are correlated to choose its indexing.
+
+This is the only point at which a model affects which observations are merged together, and it is
+not gated by the null — there is no merge yet to test the model on. It chooses only between
+indexings that are equally valid for each crystal; a model that does not belong to the crystal
+carries no information to choose with, so the crystals stay mixed, as they would without a model.
+After the merge, the model is validated as in the steps above. Without `-C` and `-S`, or on rotation
+data, the model is not used before the merge at all.
+
+## Where a model can touch the space group
+
+Spelled out, because this is the question that matters most:
+
+1. **The space group in `<prefix>.mtz` is decided without the model** — by the space-group search,
+   or by you (`-S`, `-z`). The model is read after that decision.
+2. **The model can change one thing about that group: its hand,** between the two members of an
+   enantiomorphic pair (P3₁/P3₂, P4₁2₁2/P4₃2₁2, P6₁22/P6₅22, …). The intensities are identical for
+   both, so this is the one group choice no data can make; a model that fits carries chemical
+   knowledge (the hand of its amino acids) the intensities do not. It is a label: no reflection
+   moves. It applies to a group you gave with `-S` or `-z` too. The report says
+   `SPACE_GROUP_ENANTIOMORPH= ASSUMED_FROM_MODEL` — *assumed*, not determined — also where the model
+   simply agrees with the hand the run wrote.
+3. **Everything else the model prefers goes beside the answer.** Other screw axes with the same
+   rotations (e.g. P2 against P2₁ where the absences did not decide), or the twin hypothesis'
+   subgroup: `<prefix>_mr.mtz`/`.cif`. The model never changes the point group, the lattice or the
+   centring of `<prefix>.mtz`.
+
+The safeguards behind 2 and 3:
+
+- **The null.** The model must beat its own random placements by 4.51 sd, rigid body and molecular
+  replacement alike, before it may relabel anything or have a `<prefix>_mr` file written.
+- **The anomalous veto.** The hand is taken only where the anomalous difference map, read at the
+  model's atoms, is not inverted. The veto can act only where the merge kept the Bijvoet split —
+  every rotation merge does; serial stills only with `-A`. R-free cannot see the hand at all.
+- **The indexing margin.** An indexing is taken only where the model's preference beats what a
+  randomly placed copy of it would show.
+- **Molecular replacement's own rules.** Its choice of group has its own null (`NULL_SIGMA`); a
+  placement must pack (`CLASH`, `MATTHEWS`); a choice that translational NCS could explain, or one
+  where the merge's own group was searched to fewer copies, is left `UNDECIDED`; a single domain of
+  a prediction cut by its PAE can only confirm the answer's group, never change it; in P1 the first
+  copy's origin is fixed and it counts only on its rotation function. The details are in
+  [Molecular replacement](RUGNUX_REPORT.md#molecular-replacement).
+- **The wrong-model test.** Rugnux's validation battery runs a set of datasets once without a model
+  and once with each of five models of *unrelated* proteins (similar size, unrelated sequence and
+  fold, and for the enantiomorphic sets a model in the other hand), and requires the reflection data
+  and every report key outside section 5 to be identical, `MODEL_FIT` not accepted, molecular
+  replacement not placing the model and no model decision taken
+  ("The wrong-model null" in `tools/battery/README.md`; see also [Tests](TESTS.md)).
+
+## Free set and model bias
+
+**How the free set is made.** Rugnux draws its own test set (5 %, raised towards about 500 reflections on small data but never above 10 %):
+each reflection's flag is a hash of its index, taken over its orbit under the lattice's metric
+symmetry, so symmetry mates, Friedel mates and twin-law mates share a flag, every dataset of one
+crystal form written on the same axes gets the same set, and the set does not depend on the space
+group a file is merged in
+([R-free test-set flags](CPU_DATA_ANALYSIS_INTEGRATION.md#107-r-free-test-set-flags)). The model
+plays no part in it. Where a fitting model moves the files to its setting, the set is drawn again on
+the new axes by the same rule; an indexing-ambiguity reindex leaves it unchanged, since twin-law
+mates share a flag. With `-z`, the reference's flags are inherited where the merge matches it
+(`REFERENCE_FREE_FLAGS_INHERITED=`).
+
+**What is fitted on which reflections.** Everything fitted to place and scale the model — the
+overall scale, the anisotropic *B*, the bulk solvent, the rigid body — is fitted on the working
+reflections only. The free set is then used for a few discrete *selections*: whether the rigid-body
+step is kept, which indexing wins, and in molecular replacement which placement and which group.
+Each of those leaves R-free very slightly optimistic where it fires. The coordinates, B-factors and
+occupancies of the model are never refined.
+
+**What `R_FREE` means here — and what it does not.** It is the model as placed, scored on Rugnux's
+free set: a check that the model fits these data, and a trend number between datasets of one
+campaign. It is **not** a refinement R-free, and it is not cross-validated against the model's own
+history:
+
+- A **deposited model** was refined by its depositor against *their* working set. Rugnux's free set
+  is a different 5 %, so about 95 % of Rugnux's free reflections were in the working set the model
+  was refined on. R-free here therefore reads close to R-work (the two differ by a median 0.001
+  over our corpus), and it is optimistic in the usual model-bias sense.
+- To score a deposited model on its own test set, give its structure factors as the reference
+  (`-z 1abc-sf.cif`): the depositor's free flags are then inherited. But `-z` is an assertion — it
+  also fixes the space group and the cell — so that is no longer a de-novo run.
+- A **predicted model** has not been refined against any diffraction data, so it carries no such
+  memory; what remains is the selection effect above.
+- For comparing two runs against one model, read `R_MODEL_SHELL_SCALED`, not `R_FREE`
+  ([Comparing two runs](RUGNUX_REPORT.md#comparing-two-runs-r_model_shell_scaled)).
+
+**What cannot leak, and what can.** The model never enters the merged intensities, their sigmas, the
+choice of reflections, the resolution cut or the weights, so `<prefix>.mtz` carries no model bias in
+its numbers — only the class-(b) labels. The **maps** are model-phased, and carry the model bias any
+model-phased map carries: read the mFo−DFc map, and treat density that only the 2mFo−DFc map shows
+with the usual care. CC(model, data) is read in one direction only: a confirmed shell is evidence for
+the data, a weak shell says nothing about them, and nothing acts on it.
+
+## Practical guidance
+
+**When to give a model.**
+
+- **You have a structure of this crystal form** (a fragment campaign, a ligand soak, a reference
+  structure): give it with `--model`. You get maps and a placed model ready for refinement, the
+  indexing of every dataset brought into the model's, and the hand fixed — the things that make a
+  campaign's files directly comparable. A wrong or rejected model costs a few seconds and changes
+  nothing.
+- **You have a structure of the same protein in another crystal form**: `--model` too. The rigid
+  body will not place it, and molecular replacement is tried automatically.
+- **You have only a prediction** (AlphaFold, ModelCIF): `--model` (or `--mr`). Download the PAE JSON
+  with it and keep its AlphaFold DB file name, so it is cut into domains when the whole model does
+  not place. A prediction is good for: confirming the space group, naming the hand where the data
+  cannot, offering a twinned or lower-symmetry alternative where the data are ambiguous, and a
+  first set of maps. It is not a refined model, and its R-factors read accordingly.
+- **You have a previous dataset of the same crystal form** and want the same indexing and free set:
+  `-z`, not `--model`. It is the stronger tool for the indexing ambiguity and the only one that
+  carries a free set over, but it pins the space group and cell.
+- **You want to compare with another program's statistics**: `--report-resolution`, which changes
+  nothing else.
+
+**Which file to use for what.**
+
+| File | What it is | Use it for |
+|---|---|---|
+| `<prefix>.mtz` / `.cif` / `.hkl` | The answer: the data's space group, with only the class-(b) labels from the model | Refinement, deposition, everything by default |
+| `<prefix>_model.cif` / `.pdb`, `<prefix>_2fofc.ccp4`, `_fofc.ccp4`, `_anom.ccp4`, `_maps.mtz` | The model as placed, in the frame of `<prefix>.mtz`, and its maps | Starting refinement (REFMAC5, phenix.refine), PanDDA, Coot |
+| `<prefix>_twin_<SG>.mtz` / `.cif` | The same data merged in a subgroup with a twin law — written from the data, model or not | Twin refinement, where you suspect the higher group absorbed a twin law |
+| `<prefix>_mr.mtz` / `.cif`, `<prefix>_mr_model.pdb` / `.cif`, `<prefix>_mr_*` maps | The merge in the space group the model chose (`MODEL_CHOICE_SPACE_GROUP=`), and the model placed there | Refinement *if* you accept the model's choice; compare R-free against the answer's |
+
+`<prefix>_mr` and `<prefix>_twin_<SG>` can describe the same subgroup; the `_mr` file then also
+carries the model placed in it.
+
+**Reading section 5 of the report**, in order:
+
+1. `MODEL_VALIDATION=` — did it run? If `NOT_PERFORMED`, `MODEL_VALIDATION_REASON=` says why.
+2. `MODEL_PLACEMENT=` — how the model was placed: `RIGID_BODY`, `MR` or `NOT_PLACED`.
+3. `MODEL_FIT=`, `MODEL_FIT_SIGMA=` — the rigid body's verdict against its null. Absent where the
+   rigid body was not tried (a prediction, `--mr`).
+4. `MODEL_DECISIONS_TAKEN=` — `NONE` means the reflection files are what a run without a model
+   writes. Otherwise `ENANTIOMORPH`, `INDEXING`, both, or `MR_ENANTIOMORPH`, with
+   `MODEL_ENANTIOMORPH_ADOPTED=` and `MODEL_INDEXING_OPERATOR=`.
+5. `MODEL_CHOICE_FILE_PREFIX=`, `MODEL_CHOICE_SPACE_GROUP=`, `MR_DECISION=` — whether a
+   `<prefix>_mr` file was written, and in which group.
+6. `R_WORK=`, `R_FREE=`, `CC_MODEL_CONFIRMED_TO_D_MIN=`, the anomalous sites — how well the placed
+   model describes the data (class d).
+
+Outside section 5, `SPACE_GROUP_ENANTIOMORPH=` (`UNDETERMINED`, `GIVEN` or `ASSUMED_FROM_MODEL`),
+`SETTING_SOURCE=` and the presence or absence of `INDEXING_AMBIGUITY_OPERATORS=` say what the model
+did to the answer's labels. Every key is described in [The results report](RUGNUX_REPORT.md#model-validation).
+
+## Related pages
+
+- [Advanced Rugnux](RUGNUX_ADVANCED.md#reference-data-and-the-indexing-ambiguity) — what `-z` does,
+  the indexing ambiguity, the setting the files are written in, and
+  [validating against a model](RUGNUX_ADVANCED.md#validating-against-a-model-rugnux---model).
+- [The results report](RUGNUX_REPORT.md) — [model validation](RUGNUX_REPORT.md#model-validation),
+  [molecular replacement](RUGNUX_REPORT.md#molecular-replacement), the
+  [twin hypothesis](RUGNUX_REPORT.md#the-twin-hypothesis-twin_subgroup_) and the
+  [reference-range table](RUGNUX_REPORT.md#the-reference-range-table).
+- [Data analysis: space group and validation](CPU_DATA_ANALYSIS_DECISIONS.md) — the space-group
+  search (§13) and the model-validation algorithms (§14).
+- [Reference data](CPU_DATA_ANALYSIS_INTEGRATION.md#109-reference-data-fixing-the-space-group-and-resolving-the-indexing-ambiguity)
+  and [R-free test-set flags](CPU_DATA_ANALYSIS_INTEGRATION.md#107-r-free-test-set-flags).
diff --git a/docs/index.rst b/docs/index.rst
index 214d3a34a..5b9853a3c 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -26,6 +26,7 @@ Jungfraujoch is distributed under the GPLv3 license.
    RUGNUX_INTEGRATION
    RUGNUX_REPORT
    RUGNUX_ADVANCED
+   RUGNUX_MODEL_AND_REFERENCE
    RUGNUX_CALIBRATION
    CPU_DATA_ANALYSIS
    CPU_DATA_ANALYSIS_IMAGE
diff --git a/image_puller/CMakeLists.txt b/image_puller/CMakeLists.txt
index 2ac60e680..4fb07c230 100644
--- a/image_puller/CMakeLists.txt
+++ b/image_puller/CMakeLists.txt
@@ -4,5 +4,7 @@ ADD_LIBRARY(JFJochImagePuller ZMQImagePuller.cpp ZMQImagePuller.h
         TestImagePuller.cpp
         TestImagePuller.h
         TCPImagePuller.cpp
-        TCPImagePuller.h)
+        TCPImagePuller.h
+        ReceiveRingBuffer.cpp
+        ReceiveRingBuffer.h)
 TARGET_LINK_LIBRARIES(JFJochImagePuller JFJochCommon JFJochZMQ JFJochLogger CBORStream2FrameSerialize)
diff --git a/image_puller/ImagePuller.cpp b/image_puller/ImagePuller.cpp
index 98e10f485..4855a61b0 100644
--- a/image_puller/ImagePuller.cpp
+++ b/image_puller/ImagePuller.cpp
@@ -3,7 +3,15 @@
 
 #include "ImagePuller.h"
 
-ImagePuller::ImagePuller(size_t fifo_size) : outside_fifo(fifo_size) {}
+ImagePuller::ImagePuller(size_t fifo_size, size_t ring_size) : outside_fifo(fifo_size), ring(ring_size) {}
+
+size_t ImagePuller::GetRingBufferCapacity() const {
+    return ring.GetCapacity();
+}
+
+size_t ImagePuller::GetMaxRingBufferUtilization() const {
+    return ring.GetMaxUtilization();
+}
 
 size_t ImagePuller::GetCurrentFifoUtilization() const {
     return outside_fifo.GetCurrentUtilization();
diff --git a/image_puller/ImagePuller.h b/image_puller/ImagePuller.h
index 41c8533dd..2d1d433dc 100644
--- a/image_puller/ImagePuller.h
+++ b/image_puller/ImagePuller.h
@@ -3,6 +3,7 @@
 
 #pragma once
 
+#include <atomic>
 #include <string>
 #include <memory>
 
@@ -10,6 +11,7 @@
 #include "../frame_serialize/CBORStream2Deserializer.h"
 #include "../common/ThreadSafeFIFO.h"
 #include "../common/JfjochTCP.h"
+#include "ReceiveRingBuffer.h"
 
 struct PullerAckMessage {
     TCPFrameType ack_for = TCPFrameType::DATA;
@@ -25,12 +27,14 @@ struct PullerAckMessage {
 
 struct RawFrame {
     TcpFrameHeader header{};
-    std::vector<uint8_t> payload;
     bool end = false;
 };
 
 struct ImagePullerOutput {
-    std::shared_ptr<ZMQMessage> zmq_msg;
+    // The message as it came off the socket, in the puller's ring buffer. It stays there for as long as
+    // any copy of this struct is alive - the deserialised image in cbor points into it, and republish
+    // sends it as is.
+    std::shared_ptr<ReceiveBufferRegion> raw;
     std::shared_ptr<RawFrame> tcp_msg;
     std::shared_ptr<CBORStream2DeserializerOutput> cbor;
 };
@@ -42,16 +46,32 @@ struct ImagePullerStatistics {
 class ImagePuller {
 protected:
     ThreadSafeFIFO<ImagePullerOutput> outside_fifo; // assuming 1 file is <1000 files, this should be enough to buffer
+    // Every received message is held here until consumed; the queues count messages, this bounds bytes.
+    // Size 0 means no ring: messages are held on the heap, bounded by the queues alone, as they always
+    // were - the broker's own DECTRIS puller runs that way, jfjoch_writer sets a size.
+    ReceiveRingBuffer ring;
+
+    // Bytes republish holds right now. Republish keeps the OLDEST not yet sent images, and the ring is
+    // reclaimed in arrival order, so an image waiting for a slow republish consumer pins everything
+    // received after it - the writer would then run at the republish rate. What republish may hold is
+    // therefore capped at a share of the ring; an image over the cap is not republished, exactly as one
+    // that finds the republish queue full. Without a ring there is no cap.
+    std::atomic<size_t> repub_bytes{0};
+    bool RepubCanHold(size_t bytes) const {
+        return ring.GetCapacity() == 0 || repub_bytes.load() + bytes <= ring.GetCapacity() / 4;
+    }
 public:
     constexpr static size_t DefaultQueueSize = 1000;
     constexpr static auto PollTimeout = std::chrono::milliseconds(100);
 
-    explicit ImagePuller(size_t fifo_size = DefaultQueueSize);
+    explicit ImagePuller(size_t fifo_size = DefaultQueueSize, size_t ring_size = 0);
     virtual ~ImagePuller() = default;
 
     [[nodiscard]] std::optional<ImagePullerOutput> PollImage(std::chrono::microseconds timeout = PollTimeout);
     [[nodiscard]] size_t GetMaxFifoUtilization() const;
     [[nodiscard]] size_t GetCurrentFifoUtilization() const;
+    [[nodiscard]] size_t GetRingBufferCapacity() const;
+    [[nodiscard]] size_t GetMaxRingBufferUtilization() const;
 
     virtual bool SupportsAck() const { return false; }
     virtual bool SendAck(const PullerAckMessage &) { return false; }
diff --git a/image_puller/ReceiveRingBuffer.cpp b/image_puller/ReceiveRingBuffer.cpp
new file mode 100644
index 000000000..c01b86c1e
--- /dev/null
+++ b/image_puller/ReceiveRingBuffer.cpp
@@ -0,0 +1,146 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "ReceiveRingBuffer.h"
+
+#include <algorithm>
+#include <optional>
+
+struct ReceiveRingBuffer::State {
+    // Uninitialised on purpose: a ring that is never filled should not cost its capacity in
+    // resident memory.
+    std::unique_ptr<uint8_t[]> memory;
+    const size_t capacity;
+
+    std::mutex m;
+    std::condition_variable cv;
+
+    // Live regions in arrival order. Only the oldest can give back space, so a region released out of
+    // turn is marked and waits.
+    struct Slot {
+        size_t offset;
+        size_t size;
+        bool released = false;
+    };
+    std::deque<Slot> slots;
+    // Next byte to hand out. The ring is empty when slots is empty; the oldest live byte is then
+    // slots.front().offset.
+    size_t tail = 0;
+
+    size_t utilization = 0;
+    size_t max_utilization = 0;
+    bool stopped = false;
+
+    explicit State(size_t in_capacity)
+            : memory(in_capacity > 0 ? std::make_unique_for_overwrite<uint8_t[]>(in_capacity) : nullptr),
+              capacity(in_capacity) {}
+
+    // Where a region of `size` bytes would start, or nothing if it does not fit right now. Called
+    // under the mutex.
+    std::optional<size_t> FindSpace(size_t size) const {
+        if (slots.empty())
+            return 0;
+        const size_t head = slots.front().offset;
+        // tail has come round to head: treated as full. (After a wrap the piece between the old tail
+        // and the end is free, but it is behind the live data and not worth tracking.)
+        if (tail == head)
+            return std::nullopt;
+        if (tail > head) {
+            // Live data is [head, tail): free space is behind it to the end, or before it at the start.
+            if (capacity - tail >= size)
+                return tail;
+            if (head >= size)
+                return 0;
+            return std::nullopt;
+        }
+        // Wrapped: live data is [head, capacity) + [0, tail), free space is [tail, head).
+        if (head - tail >= size)
+            return tail;
+        return std::nullopt;
+    }
+
+    void Release(size_t offset) {
+        std::unique_lock ul(m);
+        for (auto &s: slots) {
+            if (s.offset == offset) {
+                s.released = true;
+                break;
+            }
+        }
+        while (!slots.empty() && slots.front().released) {
+            utilization -= slots.front().size;
+            slots.pop_front();
+        }
+        if (slots.empty())
+            tail = 0;
+        cv.notify_all();
+    }
+
+    void ReleaseHeap(size_t size) {
+        std::unique_lock ul(m);
+        utilization -= size;
+    }
+};
+
+ReceiveRingBuffer::ReceiveRingBuffer(size_t capacity) : state(std::make_shared<State>(capacity)) {}
+
+std::shared_ptr<ReceiveBufferRegion> ReceiveRingBuffer::Allocate(size_t size) {
+    // Nothing to hold, and a zero-length slot would share its offset with the next real one.
+    if (size == 0)
+        return std::make_shared<ReceiveBufferRegion>();
+
+    std::unique_lock ul(state->m);
+
+    if (size > state->capacity) {
+        // Heap fallback: counted, not bounded.
+        state->utilization += size;
+        state->max_utilization = std::max(state->max_utilization, state->utilization);
+        auto *region = new ReceiveBufferRegion{new uint8_t[size], size};
+        return std::shared_ptr<ReceiveBufferRegion>(region, [s = state](ReceiveBufferRegion *r) {
+            delete[] r->data;
+            s->ReleaseHeap(r->size);
+            delete r;
+        });
+    }
+
+    std::optional<size_t> offset;
+    state->cv.wait(ul, [&] {
+        if (state->stopped)
+            return true;
+        offset = state->FindSpace(size);
+        return offset.has_value();
+    });
+    if (!offset)
+        return nullptr;
+
+    state->slots.push_back({*offset, size});
+    state->tail = *offset + size;
+    state->utilization += size;
+    state->max_utilization = std::max(state->max_utilization, state->utilization);
+
+    auto *region = new ReceiveBufferRegion{state->memory.get() + *offset, size};
+    return std::shared_ptr<ReceiveBufferRegion>(region, [s = state, off = *offset](ReceiveBufferRegion *r) {
+        s->Release(off);
+        delete r;
+    });
+}
+
+void ReceiveRingBuffer::Stop() {
+    std::unique_lock ul(state->m);
+    state->stopped = true;
+    state->cv.notify_all();
+}
+
+size_t ReceiveRingBuffer::GetCapacity() const {
+    return state->capacity;
+}
+
+size_t ReceiveRingBuffer::GetUtilization() const {
+    std::unique_lock ul(state->m);
+    return state->utilization;
+}
+
+size_t ReceiveRingBuffer::GetMaxUtilization() const {
+    std::unique_lock ul(state->m);
+    return state->max_utilization;
+}
diff --git a/image_puller/ReceiveRingBuffer.h b/image_puller/ReceiveRingBuffer.h
new file mode 100644
index 000000000..c1831117a
--- /dev/null
+++ b/image_puller/ReceiveRingBuffer.h
@@ -0,0 +1,52 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#pragma once
+
+#include <condition_variable>
+#include <cstdint>
+#include <deque>
+#include <memory>
+#include <mutex>
+
+// A received message lives here from the moment it comes off the socket until the last consumer is
+// done with it - the writer has stored the image and the republish thread has sent it. Both consumers
+// hold a shared_ptr to the region, and the region's destructor is what gives the bytes back.
+struct ReceiveBufferRegion {
+    uint8_t *data = nullptr;
+    size_t size = 0;
+};
+
+// Bounds the memory a puller holds in received, not yet consumed, messages - by bytes, which is the
+// quantity that matters, where the queues in front of and behind it count messages. One image of a
+// large detector is tens of megabytes, so a queue of a thousand of them is a quantity of memory nobody
+// chose.
+//
+// Regions are handed out in arrival order and the space is taken back in arrival order too, which is
+// what makes it a ring: the writer and the republish thread may finish with messages out of order, so
+// a region released early is only remembered, and the space is reclaimed once everything received
+// before it is released as well. Republish holds at most its queue's worth of messages, so that is as
+// far as the reclaim can lag.
+//
+// A message larger than the whole ring goes on the heap instead, so an oversized START or END message
+// cannot wedge the receiver; everything else waits for space, which is the back-pressure that keeps the
+// sender from running ahead.
+class ReceiveRingBuffer {
+    struct State;
+    std::shared_ptr<State> state;
+public:
+    explicit ReceiveRingBuffer(size_t capacity);
+
+    // Blocks until the bytes are available; returns nullptr once Stop() has been called. A request
+    // larger than the capacity (so every request of a ring of capacity 0) is served from the heap,
+    // at once and whether stopped or not.
+    std::shared_ptr<ReceiveBufferRegion> Allocate(size_t size);
+
+    // Releases every waiter in Allocate; the regions already handed out stay valid.
+    void Stop();
+
+    size_t GetCapacity() const;
+    // Bytes held in live regions right now / at most since construction (heap fallbacks included).
+    size_t GetUtilization() const;
+    size_t GetMaxUtilization() const;
+};
diff --git a/image_puller/TCPImagePuller.cpp b/image_puller/TCPImagePuller.cpp
index fdb5e792c..122729127 100644
--- a/image_puller/TCPImagePuller.cpp
+++ b/image_puller/TCPImagePuller.cpp
@@ -46,8 +46,10 @@ static std::pair<std::string, uint16_t> ParseTcpAddressPull(const std::string &a
 TCPImagePuller::TCPImagePuller(const std::string &tcp_addr,
                                std::optional<int32_t> rcv_buffer_size,
                                const std::string &repub_address,
-                               const std::optional<int32_t> &repub_watermark)
-    : addr(tcp_addr),
+                               const std::optional<int32_t> &repub_watermark,
+                               const std::optional<size_t> &ring_size)
+    : ImagePuller(DefaultQueueSize, ring_size.value_or(0)),
+      addr(tcp_addr),
       receive_buffer_size(rcv_buffer_size) {
     auto parsed = ParseTcpAddressPull(tcp_addr);
     host = parsed.first;
@@ -130,7 +132,7 @@ void TCPImagePuller::CBORThread() {
             if (type == TCPFrameType::CANCEL) {
                 outside_fifo.PutBlocking(ret);
             } else {
-                ret.cbor = CBORStream2Deserialize(ret.tcp_msg->payload.data(), ret.tcp_msg->payload.size());
+                ret.cbor = CBORStream2Deserialize(ret.raw ? ret.raw->data : nullptr, ret.raw ? ret.raw->size : 0);
                 outside_fifo.PutBlocking(ret);
                 // A PREFLIGHT carries a StartMessage payload but starts nothing, so republishing it
                 // would announce a run to downstream consumers that is not going to happen.
@@ -138,11 +140,13 @@ void TCPImagePuller::CBORThread() {
                     if ((ret.cbor->msg_type == CBORImageType::START)
                         || (ret.cbor->msg_type == CBORImageType::END))
                         repub_fifo.PutBlocking(ret);
-                    else
-                        repub_fifo.Put(ret);
+                    else if (RepubCanHold(ret.raw->size) && repub_fifo.Put(ret))
+                        repub_bytes += ret.raw->size;
                 }
             }
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
+            // std::exception: the deserialiser allocates a vector per per-image array of an END message,
+            // and a bad_alloc leaving this thread would end the process.
             logger.ErrorException(e);
         }
         ret = cbor_fifo.GetBlocking();
@@ -170,11 +174,13 @@ void TCPImagePuller::RepubThread() {
                     logger.Info("Republish active");
             } else {
                 if (repub_active)
-                    repub_socket->Send(ret.tcp_msg->payload.data(), ret.tcp_msg->payload.size(), true);
+                    repub_socket->Send(ret.raw->data, ret.raw->size, true);
             }
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
             logger.ErrorException(e);
         }
+        if (ret.cbor && (ret.cbor->msg_type != CBORImageType::START) && (ret.cbor->msg_type != CBORImageType::END))
+            repub_bytes -= ret.raw->size;
         ret = repub_fifo.GetBlocking();
     }
     if (repub_active)
@@ -372,14 +378,21 @@ void TCPImagePuller::ReceiverThread() {
             ImagePullerOutput out;
             out.tcp_msg = std::make_shared<RawFrame>();
             out.tcp_msg->header = frame.header;
-            out.tcp_msg->payload.resize(frame.header.payload_size);
 
-            if (frame.header.payload_size > 0
-                && !ReadExact(out.tcp_msg->payload.data(), out.tcp_msg->payload.size())) {
-                logger.Info("TCP payload read failed, reconnecting to " + addr);
-                CloseSocket();
-                std::this_thread::sleep_for(std::chrono::milliseconds(20));
-                continue;
+            if (frame.header.payload_size > 0) {
+                // Read straight into the ring (or, with no ring, onto the heap). A full ring holds the
+                // read here, the socket buffer fills, and the pusher's send blocks - its BUSY heartbeats
+                // tell it apart from a dead writer, and it gives up only after a minute in which not one
+                // image was written.
+                out.raw = ring.Allocate(frame.header.payload_size);
+                if (!out.raw)
+                    break; // Stop() - the puller is being disconnected.
+                if (!ReadExact(out.raw->data, out.raw->size)) {
+                    logger.Info("TCP payload read failed, reconnecting to " + addr);
+                    CloseSocket();
+                    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+                    continue;
+                }
             }
 
             cbor_fifo.PutBlocking(out);
@@ -438,6 +451,8 @@ void TCPImagePuller::Disconnect() {
     cbor_fifo.Stop();
     repub_fifo.Stop();
     outside_fifo.Stop();
+    // And ReceiverThread may be waiting for ring space that the stopped consumers will never free.
+    ring.Stop();
 
     if (receiver_thread.joinable())
         receiver_thread.join();
diff --git a/image_puller/TCPImagePuller.h b/image_puller/TCPImagePuller.h
index 3a0c795ed..151e931e1 100644
--- a/image_puller/TCPImagePuller.h
+++ b/image_puller/TCPImagePuller.h
@@ -53,7 +53,8 @@ public:
     explicit TCPImagePuller(const std::string &tcp_addr,
                             std::optional<int32_t> rcv_buffer_size = {},
                             const std::string &repub_address = "",
-                            const std::optional<int32_t> &repub_watermark = {});
+                            const std::optional<int32_t> &repub_watermark = {},
+                            const std::optional<size_t> &ring_size = {});
 
     ~TCPImagePuller() override;
     bool SupportsAck() const override { return true; }
diff --git a/image_puller/ZMQImagePuller.cpp b/image_puller/ZMQImagePuller.cpp
index 5078dd593..282c6fcfb 100644
--- a/image_puller/ZMQImagePuller.cpp
+++ b/image_puller/ZMQImagePuller.cpp
@@ -4,10 +4,14 @@
 #include "ZMQImagePuller.h"
 #include "../frame_serialize/CBORStream2Serializer.h"
 
+#include <cstring>
+
 ZMQImagePuller::ZMQImagePuller(const std::string &in_addr,
                                const std::string &repub_address,
     const std::optional<int32_t> &rcv_watermark,
-    const std::optional<int32_t> &repub_watermark) :
+    const std::optional<int32_t> &repub_watermark,
+    const std::optional<size_t> &ring_size) :
+        ImagePuller(DefaultQueueSize, ring_size.value_or(0)),
         socket (ZMQSocketType::Pull), addr(in_addr) {
     auto start_time = std::chrono::steady_clock::now();
     puller_thread = std::thread(&ZMQImagePuller::PullerThread, this);
@@ -43,6 +47,8 @@ void ZMQImagePuller::Disconnect() {
     cbor_fifo.Stop();
     repub_fifo.Stop();
     outside_fifo.Stop();
+    // And PullerThread may be waiting for ring space that the stopped consumers will never free.
+    ring.Stop();
 
     if (puller_thread.joinable())
         puller_thread.join();
@@ -59,8 +65,7 @@ void ZMQImagePuller::Disconnect() {
 
 void ZMQImagePuller::PullerThread() {
     while (true) {
-        ImagePullerOutput ret;
-        ret.zmq_msg = std::make_shared<ZMQMessage>();
+        auto msg = std::make_shared<ZMQMessage>();
         bool received = false;
         while (!received) {
             if (disconnect) {
@@ -68,13 +73,33 @@ void ZMQImagePuller::PullerThread() {
                 return;
             }
             try {
-                received = socket.Receive(*ret.zmq_msg, false);
+                received = socket.Receive(*msg, false);
                 if (!received)
                     std::this_thread::sleep_for(std::chrono::milliseconds(1));
             } catch (const JFJochException &e) {
                 logger.ErrorException(e);
             }
         }
+        ImagePullerOutput ret;
+        if (ring.GetCapacity() == 0) {
+            // No ring: the ZeroMQ message itself is the buffer, kept alive by the region. No copy.
+            ret.raw = std::shared_ptr<ReceiveBufferRegion>(new ReceiveBufferRegion{const_cast<uint8_t *>(msg->data()), msg->size()},
+                                                           [msg](ReceiveBufferRegion *r) { delete r; });
+        } else {
+            // Copied into the ring, and the ZeroMQ message is released with the next receive. When the
+            // ring is full this waits and the socket is not read: its queue fills to the watermark and
+            // the sender's send blocks - for as long as the sender is prepared to wait. The broker's
+            // ZeroMQ pusher waits one second, then drops the image and sends without waiting for the
+            // rest of the run, so a ring that is too small for the writer's stalls costs images.
+            ret.raw = ring.Allocate(msg->size());
+            if (!ret.raw) {
+                // Stop() - the puller is being disconnected.
+                cbor_fifo.PutBlocking(ImagePullerOutput{});
+                return;
+            }
+            if (msg->size() > 0)
+                memcpy(ret.raw->data, msg->data(), msg->size());
+        }
         cbor_fifo.PutBlocking(ret);
     }
 }
@@ -82,9 +107,9 @@ void ZMQImagePuller::PullerThread() {
 void ZMQImagePuller::CBORThread() {
     auto ret = cbor_fifo.GetBlocking();
 
-    while (ret.zmq_msg) {
+    while (ret.raw) {
         try {
-            ret.cbor = CBORStream2Deserialize(ret.zmq_msg->data(), ret.zmq_msg->size());
+            ret.cbor = CBORStream2Deserialize(ret.raw->data, ret.raw->size);
             // Even if we suspend consuming of the messages by the receiver,
             // it is still reasonable to have them republished
             // so republish functionality is not affected by Suspend()
@@ -94,10 +119,12 @@ void ZMQImagePuller::CBORThread() {
                 if ((ret.cbor->msg_type == CBORImageType::START)
                     || (ret.cbor->msg_type == CBORImageType::END))
                     repub_fifo.PutBlocking(ret);
-                else
-                    repub_fifo.Put(ret);
+                else if (RepubCanHold(ret.raw->size) && repub_fifo.Put(ret))
+                    repub_bytes += ret.raw->size;
             }
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
+            // std::exception: the deserialiser allocates a vector per per-image array of an END message,
+            // and a bad_alloc leaving this thread would end the process.
             logger.ErrorException(e);
         }
         ret = cbor_fifo.GetBlocking();
@@ -111,7 +138,7 @@ void ZMQImagePuller::RepubThread() {
     auto ret = repub_fifo.GetBlocking();
     bool repub_active = false;
 
-    while (ret.zmq_msg) {
+    while (ret.raw) {
         try {
             if (ret.cbor->msg_type == CBORImageType::START) {
                 // Start message needs to be cleaned when running republish
@@ -125,11 +152,13 @@ void ZMQImagePuller::RepubThread() {
                     logger.Info("Republish active");
             } else {
                 if (repub_active)
-                    repub_socket->Send(ret.zmq_msg->data(), ret.zmq_msg->size(), true);
+                    repub_socket->Send(ret.raw->data, ret.raw->size, true);
             }
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
             logger.ErrorException(e);
         }
+        if (ret.cbor && (ret.cbor->msg_type != CBORImageType::START) && (ret.cbor->msg_type != CBORImageType::END))
+            repub_bytes -= ret.raw->size;
         ret = repub_fifo.GetBlocking();
     }
     if (repub_active)
diff --git a/image_puller/ZMQImagePuller.h b/image_puller/ZMQImagePuller.h
index 071b5fc18..84770ab38 100644
--- a/image_puller/ZMQImagePuller.h
+++ b/image_puller/ZMQImagePuller.h
@@ -15,7 +15,9 @@
 
 
 class ZMQImagePuller : public ImagePuller {
-    const uint32_t default_receive_watermark = 100;
+    // Messages, not bytes: what the socket holds beyond the ring buffer. Kept small for that reason -
+    // the ring is where the memory is accounted for.
+    const uint32_t default_receive_watermark = 10;
     const uint32_t default_repub_watermark = 220;
 
     // ZeroMQ receive timeout allows to check for abort value from time to time
@@ -47,7 +49,8 @@ public:
             const std::string &zmq_addr,
             const std::string &repub_address = "",
             const std::optional<int32_t> &rcv_watermark = {},
-            const std::optional<int32_t> &repub_watermark = {});
+            const std::optional<int32_t> &repub_watermark = {},
+            const std::optional<size_t> &ring_size = {});
     ~ZMQImagePuller();
 
     void Disconnect() override;
diff --git a/image_pusher/ZMQStream2PusherSocket.cpp b/image_pusher/ZMQStream2PusherSocket.cpp
index 2b9334a97..64475b649 100644
--- a/image_pusher/ZMQStream2PusherSocket.cpp
+++ b/image_pusher/ZMQStream2PusherSocket.cpp
@@ -3,6 +3,8 @@
 
 #include "ZMQStream2PusherSocket.h"
 
+#include <spdlog/spdlog.h>
+
 void zmq_socket_free(void *data, void *hint) {
     auto z = (ZeroCopyReturnValue *) hint;
     z->release();
@@ -56,6 +58,10 @@ void ZMQStream2PusherSocket::WriterThread() {
         // When hitting timeout on SendZeroCopy consider it a transmission error and switch to non-blocking communication
         bool blocking_send = !transmission_error;
         if (!s.SendZeroCopy(e.z->GetImage(), e.z->GetImageSize(), zmq_socket_free, e.z, blocking_send)) {
+            if (!transmission_error)
+                spdlog::warn("ZeroMQ image socket {}: send blocked for over a second - image {} dropped, and "
+                             "images that find the queue full are dropped for the rest of the run",
+                             s.GetEndpointName(), e.z->GetImageNumber());
             transmission_error = true;
             // The slot is already released: on a failed send SendZeroCopy closes the message, and
             // closing a zmq_msg_init_data message runs its free function (zmq_socket_free).
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index bd6eb3127..6f4632394 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -57,6 +57,7 @@ ADD_EXECUTABLE(jfjoch_test
         AcquisitionCountersTest.cpp
         IndexingUnitTest.cpp
         ThreadSafeFIFOSetTest.cpp
+        ReceiveRingBufferTest.cpp
         FPGANetworkTest.cpp
         JFPedestalTest.cpp
         JFCalibrationTest.cpp
diff --git a/tests/ReceiveRingBufferTest.cpp b/tests/ReceiveRingBufferTest.cpp
new file mode 100644
index 000000000..a020242e0
--- /dev/null
+++ b/tests/ReceiveRingBufferTest.cpp
@@ -0,0 +1,129 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include <catch2/catch_all.hpp>
+#include <future>
+#include <thread>
+
+#include "../image_puller/ReceiveRingBuffer.h"
+
+using namespace std::chrono_literals;
+
+TEST_CASE("ReceiveRingBuffer_sequential_reuse", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(1024);
+    // Far more messages than fit at once: each one is released before the next is taken.
+    for (int i = 0; i < 100; i++) {
+        auto r = ring.Allocate(300);
+        REQUIRE(r);
+        REQUIRE(r->size == 300);
+        r->data[0] = 1;
+        r->data[299] = 2;
+    }
+    CHECK(ring.GetUtilization() == 0);
+    CHECK(ring.GetMaxUtilization() == 300);
+    CHECK(ring.GetCapacity() == 1024);
+}
+
+TEST_CASE("ReceiveRingBuffer_wraps_to_the_start", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(1000);
+    auto a = ring.Allocate(400);
+    auto b = ring.Allocate(400);
+    REQUIRE(a);
+    REQUIRE(b);
+    uint8_t *a_data = a->data;
+
+    // 200 bytes left at the end and nothing free at the start: a third 400 has to wait.
+    auto f = std::async(std::launch::async, [&] { return ring.Allocate(400); });
+    CHECK(f.wait_for(50ms) == std::future_status::timeout);
+
+    a.reset(); // frees [0, 400): the waiter wraps round to the start
+    auto c = f.get();
+    REQUIRE(c);
+    CHECK(c->data == a_data);
+    CHECK(ring.GetUtilization() == 800);
+    // Never two live regions on the same bytes
+    CHECK((b->data + 400 <= c->data || c->data + 400 <= b->data));
+}
+
+TEST_CASE("ReceiveRingBuffer_full_ring_blocks", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(1000);
+    auto a = ring.Allocate(500);
+    auto b = ring.Allocate(500);   // tail has come round to head
+    REQUIRE(a);
+    REQUIRE(b);
+    uint8_t *a_data = a->data;
+    auto f = std::async(std::launch::async, [&] { return ring.Allocate(1); });
+    CHECK(f.wait_for(50ms) == std::future_status::timeout);
+    a.reset();
+    auto c = f.get();
+    REQUIRE(c);
+    CHECK(c->data == a_data);
+}
+
+TEST_CASE("ReceiveRingBuffer_out_of_order_release_reclaims_in_order", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(1000);
+    auto a = ring.Allocate(500);
+    auto b = ring.Allocate(500);
+    REQUIRE(a);
+    REQUIRE(b);
+
+    b.reset(); // the newer region goes first - as republish may do
+    CHECK(ring.GetUtilization() == 1000); // its bytes are behind a's and cannot be reused yet
+
+    auto f = std::async(std::launch::async, [&] { return ring.Allocate(600); });
+    CHECK(f.wait_for(50ms) == std::future_status::timeout);
+
+    a.reset(); // now both are reclaimed
+    auto c = f.get();
+    REQUIRE(c);
+    CHECK(ring.GetUtilization() == 600);
+}
+
+TEST_CASE("ReceiveRingBuffer_oversized_message_goes_to_heap", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(100);
+    auto small = ring.Allocate(100);
+    REQUIRE(small);
+    // Does not wait for the full ring - it never fits in it
+    auto big = ring.Allocate(1000);
+    REQUIRE(big);
+    REQUIRE(big->size == 1000);
+    big->data[999] = 1;
+    CHECK(ring.GetUtilization() == 1100);
+    CHECK(ring.GetMaxUtilization() == 1100);
+    big.reset();
+    CHECK(ring.GetUtilization() == 100);
+}
+
+TEST_CASE("ReceiveRingBuffer_zero_size", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(100);
+    auto a = ring.Allocate(0);
+    auto b = ring.Allocate(100);
+    REQUIRE(a);
+    REQUIRE(b);
+    CHECK(a->size == 0);
+    a.reset();
+    CHECK(ring.GetUtilization() == 100);
+    b.reset();
+    CHECK(ring.GetUtilization() == 0);
+}
+
+TEST_CASE("ReceiveRingBuffer_stop_releases_waiter", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(100);
+    auto a = ring.Allocate(100);
+    REQUIRE(a);
+    auto f = std::async(std::launch::async, [&] { return ring.Allocate(50); });
+    CHECK(f.wait_for(50ms) == std::future_status::timeout);
+    ring.Stop();
+    CHECK(f.get() == nullptr);
+}
+
+TEST_CASE("ReceiveRingBuffer_region_outlives_ring", "[ImagePuller][Unit]") {
+    std::shared_ptr<ReceiveBufferRegion> r;
+    {
+        ReceiveRingBuffer ring(100);
+        r = ring.Allocate(50);
+    }
+    REQUIRE(r);
+    r->data[49] = 1;
+    r.reset();
+}
diff --git a/tests/StreamWriterTest.cpp b/tests/StreamWriterTest.cpp
index 662477ccf..1b767fb64 100644
--- a/tests/StreamWriterTest.cpp
+++ b/tests/StreamWriterTest.cpp
@@ -11,6 +11,7 @@
 #include "../image_pusher/ZMQStream2Pusher.h"
 #include "../receiver/JFJochReceiverService.h"
 #include "../image_pusher/ZMQWriterNotificationPuller.h"
+#include "../image_puller/TestImagePuller.h"
 
 TEST_CASE("StreamWriterTest_ZMQ", "[StreamWriter]") {
     RegisterHDF5Filter();
@@ -246,3 +247,52 @@ TEST_CASE("StreamWriterTest_ZMQ_MasterAfterAllDataFiles", "[StreamWriter]") {
 
     std::filesystem::remove_all("subdir4");
 }
+
+// A failure that is not a JFJochException must end the run the same way as one that is: error state,
+// END processed, state Finalized. Before this test the writer caught only its own exception type at the
+// per-image and END sites, and anything else - here the JSON library refusing a sample name that is not
+// UTF-8 while publishing a file-done notification - left Run() with the state stuck in Receiving, no END
+// acknowledgement and no notification, and on TCP a broker waiting for that acknowledgement for ever.
+TEST_CASE("StreamWriterTest_NonJFJochException_EndsRunInErrorState", "[StreamWriter]") {
+    RegisterHDF5Filter();
+    Logger logger("StreamWriterTest_NonJFJochException");
+
+    DiffractionExperiment x(DetJF(2));
+    x.FilePrefix("subdir5/StreamWriterTest5").NumTriggers(1).ImagesPerTrigger(2).ImagesPerFile(1)
+            .Compression(CompressionAlgorithm::NO_COMPRESSION).OverwriteExistingFiles(true)
+            .SampleName("\xff\xfe not utf-8");
+
+    StartMessage start_message;
+    x.FillMessage(start_message);
+    start_message.write_master_file = true;
+
+    TestImagePuller puller;
+    // The file-done PUB socket is what makes the writer serialise the sample name to JSON.
+    StreamWriter writer(logger, puller, "ipc://*");
+
+    puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(start_message)});
+    std::vector<int16_t> image(x.GetPixelsNum());
+    for (int i = 0; i < 2; i++) {
+        DataMessage message{};
+        message.image = CompressedImage(image, x.GetXPixelsNum(), x.GetYPixelsNum());
+        message.number = i;
+        puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(message)});
+    }
+    EndMessage end_message{};
+    end_message.max_image_number = 2;
+    puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(end_message)});
+
+    StreamWriterOutput out;
+    REQUIRE_NOTHROW(out = writer.Run());
+
+    // Image 0 filled its one-image data file; closing it published the notification, which threw.
+    CHECK(out.stats.state == StreamWriterState::Finalized);
+    CHECK(out.stats.processed_images == 0);
+    CHECK(out.data_file_stats.empty());
+    // The master is never written for a run that ended in error; the data file was renamed into
+    // place before the notification was attempted.
+    CHECK(!std::filesystem::exists("subdir5/StreamWriterTest5_master.h5"));
+    CHECK(std::filesystem::exists("subdir5/StreamWriterTest5_data_000001.h5"));
+
+    std::filesystem::remove_all("subdir5");
+}
diff --git a/tests/TCPImagePusherTest.cpp b/tests/TCPImagePusherTest.cpp
index 19b5b0ae9..d19a67a80 100644
--- a/tests/TCPImagePusherTest.cpp
+++ b/tests/TCPImagePusherTest.cpp
@@ -41,8 +41,13 @@ TEST_CASE("TCPImageCommTest_2Writers_WithAck", "[TCP]") {
 
     TCPStreamPusher pusher("tcp://127.0.0.1:*", npullers);
     std::vector<std::unique_ptr<TCPImagePuller> > puller;
+    // The second writer receives into a ring that holds three images (with their CBOR framing): a
+    // full ring stalls its read, the pusher's send blocks, and nothing may be lost or reordered.
+    const size_t small_ring = 3 * x.GetPixelsNum() * sizeof(uint16_t) + 3 * 65536;
     for (int i = 0; i < npullers; i++)
-        puller.push_back(std::make_unique<TCPImagePuller>(pusher.GetAddress()[0], 64 * 1024 * 1024));
+        puller.push_back(std::make_unique<TCPImagePuller>(pusher.GetAddress()[0], 64 * 1024 * 1024, "", std::nullopt,
+                                                          (i == 1) ? std::optional<size_t>(small_ring) : std::nullopt));
+    REQUIRE(puller[1]->GetRingBufferCapacity() == small_ring);
 
     // Wait for all pullers to connect before starting data collection
     for (int attempt = 0; attempt < 100 && pusher.GetConnectedWriters() < static_cast<size_t>(npullers); ++attempt)
diff --git a/tests/ZMQImagePusherTest.cpp b/tests/ZMQImagePusherTest.cpp
index beaa18517..f6e541e52 100644
--- a/tests/ZMQImagePusherTest.cpp
+++ b/tests/ZMQImagePusherTest.cpp
@@ -420,3 +420,75 @@ TEST_CASE("ZMQImagePuller_RecoversFromFullQueue","[ZeroMQ]") {
 
     puller.Disconnect();
 }
+
+// The whole run goes through a ring that holds three images: the puller has to wait for space, the
+// socket queue fills, the pusher blocks, and nothing is lost or reordered. The peak ring use proves
+// the bound held.
+TEST_CASE("ZMQImageCommTest_SmallRingBuffer","[ZeroMQ]") {
+    const size_t nframes = 64;
+
+    Logger logger(Catch::getResultCapture().getCurrentTestName());
+
+    DiffractionExperiment x(DetJF(1));
+    x.Raw();
+    x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).IncidentEnergy_keV(12.4)
+            .ImagesPerTrigger(nframes).Compression(CompressionAlgorithm::NO_COMPRESSION);
+    REQUIRE(x.GetImageNum() == nframes);
+
+    std::mt19937 g1(1387);
+    std::uniform_int_distribution<uint16_t> dist;
+    std::vector<uint16_t> image1(x.GetPixelsNum()*nframes);
+    for (auto &i: image1) i = dist(g1);
+
+    const size_t image_bytes = x.GetPixelsNum() * sizeof(uint16_t);
+    const size_t ring_size = 3 * image_bytes + 3 * 65536; // three images with their CBOR framing
+
+    ZMQStream2Pusher pusher({"ipc://*"});
+    std::vector<size_t> diff_size(1), diff_content(1), diff_split(1), nimages(1);
+
+    auto pusher_addr = pusher.GetAddress();
+    ZMQImagePuller puller(pusher_addr[0], "", {}, {}, ring_size);
+    REQUIRE(puller.GetRingBufferCapacity() == ring_size);
+
+    std::thread sender_thread = std::thread([&] {
+        std::vector<uint8_t> serialization_buffer(16*1024*1024);
+        CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
+
+        StartMessage message {
+                .images_per_file = 16,
+                .write_master_file = true
+        };
+        EndMessage end_message{};
+
+        pusher.StartDataCollection(message);
+        for (int i = 0; i < nframes; i++) {
+            DataMessage data_message;
+            data_message.number = i;
+            data_message.image = CompressedImage(image1.data() + i * x.GetPixelsNum(),
+                                                 image_bytes,
+                                                 x.GetXPixelsNum(),
+                                                 x.GetYPixelsNum(),
+                                                 x.GetImageMode(),
+                                                 x.GetCompressionAlgorithm());
+            serializer.SerializeImage(data_message);
+            pusher.SendImage(serialization_buffer.data(), serializer.GetBufferSize(), i);
+        }
+
+        pusher.EndDataCollection(end_message);
+    });
+
+    std::thread puller_thread(test_puller, &puller, std::cref(x), std::cref(image1), 1, 0,
+                              std::ref(diff_split), std::ref(diff_size), std::ref(diff_content),
+                              std::ref(nimages));
+
+    sender_thread.join();
+    puller_thread.join();
+
+    puller.Disconnect();
+
+    REQUIRE(nimages[0] == nframes);
+    REQUIRE(diff_size[0] == 0);
+    REQUIRE(diff_content[0] == 0);
+    CHECK(puller.GetMaxRingBufferUtilization() <= ring_size);
+    CHECK(puller.GetMaxRingBufferUtilization() >= image_bytes);
+}
diff --git a/tools/battery/inhouse.json b/tools/battery/inhouse.json
index e796742f0..dcc8bc1d7 100644
--- a/tools/battery/inhouse.json
+++ b/tools/battery/inhouse.json
@@ -27,9 +27,9 @@
  {"id": "insu_I_x06da_5keV", "input": "insu_I_x06da_5keV/Ins_I_6_006_master.h5", "wavelength": 2.47923, "ref": {"sgno": 197, "cell": [77.785, 77.785, 77.785, 90.0, 90.0, 90.0], "anomalous": true, "isa": 17.54, "completeness": 91.7, "r_meas": 0.073, "cc_half": 0.999, "multiplicity": 11.17, "dmin_low": 7.21, "r_meas_low": 0.048, "cc_half_high": 0.855, "dmin": 2.45, "dmin_rule": "xds_range", "dmin_xds": 2.45, "dmax": 50.0}, "tags": ["h5", "insulin", "long-wavelength"], "model": "2bn3"},
  {"id": "insu_I_x06da_6keV", "input": "insu_I_x06da_6keV/Ins_I_6_004_master.h5", "wavelength": 2.06648, "ref": {"sgno": 197, "cell": [77.71, 77.71, 77.71, 90.0, 90.0, 90.0], "anomalous": true, "isa": 17.95, "completeness": 92.4, "r_meas": 0.065, "cc_half": 0.999, "multiplicity": 12.42, "dmin_low": 6.03, "r_meas_low": 0.047, "cc_half_high": 0.905, "dmin": 2.04, "dmin_rule": "xds_range", "dmin_xds": 2.04, "dmax": 50.0}, "tags": ["h5", "insulin", "long-wavelength"], "model": "2bn3"},
  {"id": "insu_I_x06da_5keV_2", "input": "insu_I_x06da_5keV_2/Ins_I_7_001_master.h5", "wavelength": 2.47923, "ref": {"sgno": 197, "cell": [77.694, 77.694, 77.694, 90.0, 90.0, 90.0], "anomalous": true, "isa": 20.01, "completeness": 91.8, "r_meas": 0.076, "cc_half": 0.999, "multiplicity": 12.73, "dmin_low": 7.21, "r_meas_low": 0.048, "cc_half_high": 0.888, "dmin": 2.45, "dmin_rule": "xds_range", "dmin_xds": 2.45, "dmax": 50.0}, "tags": ["h5", "insulin", "long-wavelength"], "tiers": {"smoke": "cubic insulin at 5 keV (long wavelength), fast"}, "model": "2bn3"},
- {"id": "cytc_x10sa", "input": "cytc_x10sa/cytC_10_002_master.h5", "wavelength": 0.99989, "ref": {"sgno": 152, "cell": [83.659, 83.659, 86.011, 90.0, 90.0, 120.0], "anomalous": true, "isa": 31.75, "completeness": 99.8, "r_meas": 0.232, "cc_half": 0.999, "multiplicity": 10.65, "dmin_low": 6.05, "r_meas_low": 0.036, "cc_half_high": 0.046, "dmin": 2.267, "dmin_rule": "cc_half_0.30", "dmin_xds": 2.039, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif", "tiers": {"ci": "trigonal P3121 with tNCS; the hand comes from the reference model"}},
- {"id": "cytc_x06da_1", "input": "cytc_x06da_1/cytC_2_0770b5_master.h5", "wavelength": 0.95375, "ref": {"sgno": 152, "cell": [83.836, 83.836, 86.733, 90.0, 90.0, 120.0], "anomalous": true, "isa": 24.49, "completeness": 99.9, "r_meas": 0.084, "cc_half": 0.999, "multiplicity": 7.67, "dmin_low": 5.58, "r_meas_low": 0.033, "cc_half_high": 0.366, "dmin": 1.875, "dmin_rule": "xds_range", "dmin_xds": 1.875, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif"},
- {"id": "cytc_x06da_2", "input": "cytc_x06da_2/cytC_3_874285_master.h5", "wavelength": 0.95375, "ref": {"sgno": 152, "cell": [83.719, 83.719, 88.617, 90.0, 90.0, 120.0], "anomalous": true, "isa": 26.98, "completeness": 99.9, "r_meas": 0.097, "cc_half": 0.999, "multiplicity": 7.82, "dmin_low": 5.03, "r_meas_low": 0.032, "cc_half_high": 0.236, "dmin": 1.69, "dmin_rule": "xds_range", "dmin_xds": 1.69, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif"},
+ {"id": "cytc_x10sa", "input": "cytc_x10sa/cytC_10_002_master.h5", "wavelength": 0.99989, "ref": {"sgno": 152, "cell": [83.659, 83.659, 86.011, 90.0, 90.0, 120.0], "anomalous": true, "isa": 31.75, "completeness": 99.8, "r_meas": 0.232, "cc_half": 0.999, "multiplicity": 10.65, "dmin_low": 6.05, "r_meas_low": 0.036, "cc_half_high": 0.046, "dmin": 2.267, "dmin_rule": "cc_half_0.30", "dmin_xds": 2.039, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif", "tiers": {"ci": "trigonal P3121 with tNCS; the hand comes from the reference model"}, "ref_override": {"sg": "P 32 2 1", "sgno": 154}, "ref_override_why": "The reference model (P 32 2 1, three copies) fixes the hand that XDS left arbitrary: the 2-fold of P 32 2 1 holds to measurement noise in every shell and a P 32 six-copy model, twinned or not, fits no better on the same data (in-house analysis 2026-10-13)"},
+ {"id": "cytc_x06da_1", "input": "cytc_x06da_1/cytC_2_0770b5_master.h5", "wavelength": 0.95375, "ref": {"sgno": 152, "cell": [83.836, 83.836, 86.733, 90.0, 90.0, 120.0], "anomalous": true, "isa": 24.49, "completeness": 99.9, "r_meas": 0.084, "cc_half": 0.999, "multiplicity": 7.67, "dmin_low": 5.58, "r_meas_low": 0.033, "cc_half_high": 0.366, "dmin": 1.875, "dmin_rule": "xds_range", "dmin_xds": 1.875, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif", "ref_override": {"sg": "P 32 2 1", "sgno": 154}, "ref_override_why": "The reference model (P 32 2 1, three copies) fixes the hand that XDS left arbitrary: the 2-fold of P 32 2 1 holds to measurement noise in every shell and a P 32 six-copy model, twinned or not, fits no better on the same data (in-house analysis 2026-10-13)"},
+ {"id": "cytc_x06da_2", "input": "cytc_x06da_2/cytC_3_874285_master.h5", "wavelength": 0.95375, "ref": {"sgno": 152, "cell": [83.719, 83.719, 88.617, 90.0, 90.0, 120.0], "anomalous": true, "isa": 26.98, "completeness": 99.9, "r_meas": 0.097, "cc_half": 0.999, "multiplicity": 7.82, "dmin_low": 5.03, "r_meas_low": 0.032, "cc_half_high": 0.236, "dmin": 1.69, "dmin_rule": "xds_range", "dmin_xds": 1.69, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif", "ref_override": {"sg": "P 32 2 1", "sgno": 154}, "ref_override_why": "The reference model (P 32 2 1, three copies) fixes the hand that XDS left arbitrary: the 2-fold of P 32 2 1 holds to measurement noise in every shell and a P 32 six-copy model, twinned or not, fits no better on the same data (in-house analysis 2026-10-13)"},
  {"id": "myob_x10sa", "input": "myob_x10sa/MyoB_13_110ee0_master.h5", "wavelength": 0.99988, "ref": {"sgno": 4, "cell": [35.229, 28.461, 63.076, 90.0, 106.504, 90.0], "anomalous": true, "isa": 5.21, "completeness": 97.7, "r_meas": 0.247, "cc_half": 0.965, "multiplicity": 3.29, "dmin_low": 5.16, "r_meas_low": 0.186, "cc_half_high": 0.359, "dmin": 1.742, "dmin_rule": "xds_range", "dmin_xds": 1.742, "dmax": 50.0}, "tags": ["h5", "myoglobin"], "model": "4dc8", "tiers": {"ci": "monoclinic P21, Eiger"}},
  {"id": "myob_x06da", "input": "myob_x06da/MyoB_9_446b70_master.h5", "wavelength": 0.95365, "ref": {"sgno": 3, "cell": [35.182, 28.544, 62.83, 90.0, 106.138, 90.0], "anomalous": true, "isa": 7.59, "completeness": 99.4, "r_meas": 0.232, "cc_half": 0.987, "multiplicity": 3.47, "dmin_low": 4.23, "r_meas_low": 0.105, "cc_half_high": 0.267, "dmin": 1.422, "dmin_rule": "xds_range", "dmin_xds": 1.422, "dmax": 50.0}, "tags": ["h5", "myoglobin"], "tiers": {"smoke": "monoclinic myoglobin, screw not arbitrable by XDS"}, "ref_override": {"sg": "P 1 21 1", "sgno": 4}, "ref_override_why": "myoglobin, P 1 21 1. XDS does not judge screw axes (it reports P 2)", "model": "4dc8"},
  {"id": "myob_x06da_powder_1", "input": "myob_x06da_powder_1/MyoB2-1_75979a_master.h5", "wavelength": 0.95373, "ref": {"sgno": 1, "cell": [28.705, 34.796, 64.373, 74.314, 89.879, 90.057], "anomalous": false, "isa": 5.54, "completeness": 74.2, "r_meas": 0.493, "cc_half": 0.966, "multiplicity": 2.87, "dmin_low": 4.44, "r_meas_low": 0.125, "cc_half_high": 0.953, "dmin": 1.495, "dmin_rule": "xds_range", "dmin_xds": 1.495, "dmax": 50.0}, "tags": ["h5", "myoglobin", "twin"], "tiers": {"smoke": "known hard: triclinic twin"}, "ref_override": {"sg": "P 1 21 1", "sgno": 4, "cell": [35.14, 28.88, 64.64, 90.0, 105.9, 90.0]}, "ref_override_why": "myoglobin, P 1 21 1. XDS indexed this sweep (a powder-contaminated crystal) in P 1 on a reduced cell (28.7 34.8 64.4 74.3 89.9 90.1); the monoclinic cell is that cell's conventional setting", "model": "4dc8"},
diff --git a/tools/battery/open.json b/tools/battery/open.json
index 0fd30974c..fdc4e693f 100644
--- a/tools/battery/open.json
+++ b/tools/battery/open.json
@@ -41,11 +41,11 @@
  {"id": "6oel", "input": "6oel/1449-8_3_001.img", "wavelength": 0.99994, "ref": {"sg": "F 41 3 2", "sgno": 210, "cell": [328.1, 328.1, 328.1, 90.0, 90.0, 90.0], "dmin": 3.1}, "tags": ["SMV", "cubic"], "tiers": {"smoke": "SMV reader (ADSC), cubic F4132"}},
  {"id": "6p8p", "input": "6p8p/14_12_1_0001.cbf", "wavelength": 0.9792, "ref": {"sg": "P 4", "sgno": 75, "cell": [97.525, 97.525, 60.134, 90.0, 90.0, 90.0], "dmin": 1.635}, "tags": ["cbf", "tetragonal"]},
  {"id": "6pb3", "input": "6pb3/14_16_1_000001.cbf", "wavelength": 0.97918, "ref": {"sg": "P 6", "sgno": 168, "cell": [100.436, 100.436, 48.86, 90.0, 90.0, 120.0], "dmin": 2.048}, "tags": ["cbf", "hexagonal"]},
- {"id": "6pxb", "input": "6pxb/TJB1_3_rachel_1_0001.cbf", "wavelength": 0.9792, "pinned": true, "ref": {"sg": "P 32", "sgno": 145, "cell": [64.021, 64.021, 119.447, 90.0, 90.0, 120.0], "dmin": 1.747}, "ref_alternatives": [{"sg": "P 32 1 2", "sgno": 153, "why": "The deposition merged in point group 3 (its 55502 unique reflections to 1.747 A are what Laue class -3 holds, twice what -3 1 m would) and refined six chains in P 32. The intensities read point group 312 instead: the three added two-folds correlate at 0.98-0.99, at or above the three-folds nobody disputes (0.98), against 0.37-0.40 for the 321 and 622 operators, POINTLESS on our P1 merge picks P -3 1 m (likelihood 1.000), and the reflections centric in 312 but not in 3 are distributed as centric (+435 nats), which a twin law cannot produce. Against that, the deposited chains pair under the added two-fold at 0.3-0.7 A, more than coordinate error at 1.75 A, and the model tells the two indexings apart (R 0.22 against 0.25), so the two-fold may be a near-exact non-crystallographic one; ZANUDA settles on P 32 2 1, whose operators these data do not support. Neither answer is established, so both are accepted; P 31 1 2, which the data cannot separate from P 32 1 2, is accepted as its hand"}], "symmetry_evidence": "symmetry/6pxb", "tags": ["cbf", "trigonal"]},
+ {"id": "6pxb", "input": "6pxb/TJB1_3_rachel_1_0001.cbf", "wavelength": 0.9792, "pinned": true, "ref": {"sg": "P 32", "sgno": 145, "cell": [64.021, 64.021, 119.447, 90.0, 90.0, 120.0], "dmin": 1.747}, "ref_alternatives": [{"sg": "P 32 1 2", "sgno": 153, "why": "The deposition merged in point group 3 (its 55502 unique reflections to 1.747 A are what Laue class -3 holds, twice what -3 1 m would) and refined six chains in P 32. The intensities read point group 312 instead: the three added two-folds correlate at 0.98-0.99, at or above the three-folds nobody disputes (0.98), against 0.37-0.40 for the 321 and 622 operators, POINTLESS on our P1 merge picks P -3 1 m (likelihood 1.000), and the reflections centric in 312 but not in 3 are distributed as centric (+435 nats), which a twin law cannot produce. Against that, the deposited chains pair under the added two-fold at 0.3-0.7 A, more than coordinate error at 1.75 A, and the model tells the two indexings apart (R 0.22 against 0.25), so the two-fold may be a near-exact non-crystallographic one; ZANUDA 1.097 with the deposited model keeps P 32 on our data (three free sets) and the depositor's; the model refined in P 32 1 2 reaches R-free 0.49-0.53 against 0.33-0.37. Neither answer is established, so both are accepted; P 31 1 2, which the data cannot separate from P 32 1 2, is accepted as its hand"}], "symmetry_evidence": "symmetry/6pxb", "tags": ["cbf", "trigonal"]},
  {"id": "6pxc", "input": "6pxc/TJB6_1_Rachel_1_0001.cbf", "wavelength": 0.9791, "ref": {"sg": "I 2 2 2", "sgno": 23, "cell": [44.19, 64.827, 87.239, 90.0, 90.0, 90.0], "dmin": 1.6}, "tags": ["cbf", "orthorhombic"], "tiers": {"ci": "body-centred orthorhombic I222"}},
  {"id": "6r72", "input": "6r72/V-CK63-8-ld_1_master.h5", "wavelength": 0.98011, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [117.805, 110.754, 155.615, 90.0, 93.23, 90.0], "dmin": 3.95}, "pinned": true, "tags": ["h5", "monoclinic"]},
  {"id": "6rlr", "input": "6rlr/_b4_1_master.h5", "wavelength": 0.97949, "ref": {"sg": "P 1", "sgno": 1, "cell": [39.986, 39.998, 63.643, 80.39, 76.29, 68.15], "dmin": 2.0}, "tags": ["h5", "triclinic", "twin"]},
- {"id": "6toc", "input": "6toc/zenodo/dts_1_00001.cbf", "wavelength": 1.00004, "ref": {"sg": "P 42", "sgno": 77, "cell": [31.523, 31.523, 81.599, 90.0, 90.0, 90.0], "dmin": 1.853}, "ref_alternatives": [{"sg": "P 42 2 2", "sgno": 93, "why": "Our reduction and an independent POINTLESS run on our own P1 merge both read point group 422, and the deposited asymmetric unit's two chains are related by the very two-fold the higher group adds, to 0.16 A CA RMSD over 43 residues - coordinate error at 1.85 A. Merging in P 42 2 2 costs 0.0006 in R_meas for 1.75x the multiplicity and correlates better with the deposited model (0.9802 vs 0.9764). The refinement test is NOT unanimous: ZANUDA 1.097 refines P 42 2 2 to R-free 0.2561 against P 42's 0.2636 at half the parameters and reports the deposited assignment incorrect, while an independent Refmac 5.8.0431 comparison on a symmetry-consistent free set puts P 42 ahead by 0.004-0.020 depending on cycle count - less than the spread between refinement protocols. Neither answer is established: a pseudo-symmetry too exact for any test we have remains a live explanation, and so does the deposited assignment. Either is accepted"}], "symmetry_evidence": "symmetry/6toc", "tags": ["cbf", "tetragonal", "twin"]},
+ {"id": "6toc", "input": "6toc/zenodo/dts_1_00001.cbf", "wavelength": 1.00004, "ref": {"sg": "P 42", "sgno": 77, "cell": [31.523, 31.523, 81.599, 90.0, 90.0, 90.0], "dmin": 1.853}, "ref_alternatives": [{"sg": "P 42 2 2", "sgno": 93, "why": "Our reduction and an independent POINTLESS run on our own P1 merge both read point group 422, and the deposited asymmetric unit's two chains are related by the very two-fold the higher group adds, to 0.16 A CA RMSD over 43 residues - coordinate error at 1.85 A. Merging in P 42 2 2 costs 0.0006 in R_meas for 1.75x the multiplicity and correlates better with the deposited model (0.9802 vs 0.9764). The refinement test is NOT unanimous: ZANUDA 1.097 prefers P 42 2 2 on all four data sets tried (ours with three free sets, the depositor's), but R-free favours P 42 2 2 once by 0.008 and P 42 three times by 0.015-0.018, while an independent Refmac 5.8.0431 comparison on a symmetry-consistent free set puts P 42 ahead by 0.004-0.020 depending on cycle count - less than the spread between refinement protocols. Neither answer is established: a pseudo-symmetry too exact for any test we have remains a live explanation, and so does the deposited assignment. Either is accepted"}], "symmetry_evidence": "symmetry/6toc", "tags": ["cbf", "tetragonal", "twin"]},
  {"id": "6ttn", "input": "6ttn/6ttn/data/H6H_33_01_hyo_full_1_0001.cbf", "wavelength": 0.9184, "ref": {"sg": "P 21 21 21", "sgno": 19, "cell": [39.894, 79.841, 104.701, 90.0, 90.0, 90.0], "dmin": 1.12}, "tags": ["cbf", "orthorhombic"]},
  {"id": "6u7g", "input": "6u7g/6u7g/data/SNB02_11_8_1_master.h5", "wavelength": 1.03318, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [99.555, 98.682, 147.525, 90.0, 104.6, 90.0], "dmin": 2.35}, "pinned": true, "tags": ["h5", "monoclinic"]},
  {"id": "6ukf", "input": "6ukf/6ukf/data/XDC-7_Pn6_000001.cbf", "wavelength": 0.8, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [61.018, 37.317, 69.027, 90.0, 109.768, 90.0], "dmin": 1.0}, "tags": ["cbf", "monoclinic"]},
@@ -106,9 +106,9 @@
  {"id": "8u0i", "input": "8u0i/8u0i/data/dI-2-05a_4_00001.cbf", "wavelength": 1.00004, "ref": {"sg": "P 43 21 2", "sgno": 96, "cell": [50.317, 50.317, 90.596, 90.0, 90.0, 90.0], "dmin": 1.541}, "tags": ["cbf", "tetragonal"], "tiers": {"ci": "miniCBF, tetragonal P43212"}},
  {"id": "8v4o", "input": "8v4o/CaalA_00629_a_FS11-AMP_8v4o/data/PSL-1602_2180_master.h5", "wavelength": 0.97949, "ref": {"sg": "P 61 2 2", "sgno": 178, "cell": [139.46, 139.46, 544.99, 90.0, 90.0, 120.0], "dmin": 2.7}, "tags": ["h5", "hexagonal"]},
  {"id": "8xbp", "input": "8xbp/8xbp/data/puck2872_3_1_master.h5", "wavelength": 0.97856, "ref": {"sg": "C 1 2 1", "sgno": 5, "cell": [148.29, 50.78, 60.21, 90.0, 92.33, 90.0], "dmin": 1.99}, "ref_override": {"cell": [148.736, 51.777, 60.181, 90.0, 92.36, 90.0]}, "ref_override_why": "the uploaded sweep is a different crystal from the one the deposited cell describes: master data_collection_date 2023-06-21, deposited _diffrn_detector.pdbx_collection_date 2023-06-23, and the deposited b = 50.78 is 2.0% from the b these images give. The override is the cell DIALS 3.29 indexes de novo on this master, b = 51.777(11); the deposited space group and d_min are kept", "tags": ["h5", "monoclinic"]},
- {"id": "8xte", "input": "8xte/1101/G2_1_00001.cbf", "wavelength": 0.97853, "ref": {"sg": "P 32", "sgno": 145, "cell": [208.77, 208.77, 67.22, 90.0, 90.0, 120.0], "dmin": 1.99}, "ref_alternatives": [{"sg": "P 31 2 1", "sgno": 152, "why": "The evidence favours the higher group here. The twin-immune centric zone of the added two-folds - reflections the higher group makes centric, which are their own twin mates and so cannot be made to read centric by a merohedral twin law - gives <|E^2-1|> = 0.946 +/- 0.012 against a centric expectation of 0.968 and an acentric 0.736, at +707 nats. Re-refinement on a shared free set with the twin law removed from both sides favours P 3_2 2 1 (0.2177/0.2322) over P 3_2 (0.2460/0.2612), and the deposited entry's published R values reproduce only with an undeclared twin law h,-h-k,-l at alpha = 0.50, which is itself a 321-symmetric target. No refinement R can close the question in principle, because a merohedral twin at exactly alpha = 0.5 and true 321 predict identical intensities; the case rests on the centric zone. Both answers are accepted, ours being the better supported"}], "symmetry_evidence": "symmetry/8xte", "tags": ["cbf", "trigonal"]},
+ {"id": "8xte", "input": "8xte/1101/G2_1_00001.cbf", "wavelength": 0.97853, "ref": {"sg": "P 32", "sgno": 145, "cell": [208.77, 208.77, 67.22, 90.0, 90.0, 120.0], "dmin": 1.99}, "ref_alternatives": [{"sg": "P 31 2 1", "sgno": 152, "why": "The evidence favours the higher group here. The twin-immune centric zone of the added two-folds - reflections the higher group makes centric, which are their own twin mates and so cannot be made to read centric by a merohedral twin law - gives <|E^2-1|> = 0.946 +/- 0.012 against a centric expectation of 0.968 and an acentric 0.736, at +707 nats. Re-refinement on a shared free set with the twin law removed from both sides favours P 3_2 2 1 (0.2177/0.2322) over P 3_2 (0.2460/0.2612), and the deposited entry's published R values reproduce only with an undeclared twin law h,-h-k,-l at alpha = 0.50, which is itself a 321-symmetric target. No refinement R can close the question in principle, because a merohedral twin at exactly alpha = 0.5 and true 321 predict identical intensities; the case rests on the centric zone. ZANUDA 1.097 keeps P 32: the deposited chains are 6.1 A from the 321 symmetry and refine to R-free 0.54-0.57 in P 32 2 1; the deposited model gives the higher group no support. Both answers are accepted, neither established"}], "symmetry_evidence": "symmetry/8xte", "tags": ["cbf", "trigonal"]},
  {"id": "8xtf", "input": "8xtf/1102/puck03_12_1_master.h5", "wavelength": 0.97918, "ref": {"sg": "H 3 2", "sgno": 155, "cell": [211.75, 211.75, 67.42, 90.0, 90.0, 120.0], "dmin": 2.13}, "tags": ["h5", "trigonal"]},
- {"id": "8xtg", "input": "8xtg/1100/mpag15_1_00001.cbf", "wavelength": 0.97853, "ref": {"sg": "P 32", "sgno": 145, "cell": [199.54, 199.54, 67.15, 90.0, 90.0, 120.0], "dmin": 2.0}, "ref_alternatives": [{"sg": "P 31 2 1", "sgno": 152, "why": "The evidence favours the DEPOSITION here, and this row must not be read as the 8xte one. Every correlation-based instrument we have - our own operator correlations, POINTLESS (0.85 on our P1 merge) - reads point group 321, but our own twin-immune centric-zone test, the only one that separates real symmetry from pseudo-symmetry, read <|E^2-1|> = 0.869 at -44.9 nats against the promotion when it assumed an untwinned zone; read at the fraction of the crystal's other twin laws (rc174) it now favours the promotion, with an L-test twin fraction of 0.20-0.26. Whether this crystal is partially twinned or purely pseudo-symmetric is not established. Both answers are accepted, the deposition being the better supported"}], "symmetry_evidence": "symmetry/8xtg", "tags": ["cbf", "trigonal"]},
+ {"id": "8xtg", "input": "8xtg/1100/mpag15_1_00001.cbf", "wavelength": 0.97853, "ref": {"sg": "P 32", "sgno": 145, "cell": [199.54, 199.54, 67.15, 90.0, 90.0, 120.0], "dmin": 2.0}, "ref_alternatives": [{"sg": "P 31 2 1", "sgno": 152, "why": "The evidence favours the DEPOSITION here, and this row must not be read as the 8xte one. Every correlation-based instrument we have - our own operator correlations, POINTLESS (0.85 on our P1 merge) - reads point group 321, but our own twin-immune centric-zone test, the only one that separates real symmetry from pseudo-symmetry, read <|E^2-1|> = 0.869 at -44.9 nats against the promotion when it assumed an untwinned zone; read at the fraction of the crystal's other twin laws (rc174) it now favours the promotion, with an L-test twin fraction of 0.20-0.26. Whether this crystal is partially twinned or purely pseudo-symmetric is not established. ZANUDA 1.097 prefers P 32 2 1 on all four data sets (model within 1.2 A of it), while R-free favours P 32 by 0.006-0.019 on each. Both answers are accepted, the deposition being the better supported"}], "symmetry_evidence": "symmetry/8xtg", "tags": ["cbf", "trigonal"]},
  {"id": "8y74", "input": "8y74/1_1/1_1_master.h5", "wavelength": 0.97918, "ref": {"sg": "C 1 2 1", "sgno": 5, "cell": [125.782, 76.553, 87.143, 90.0, 92.449, 90.0], "dmin": 1.9}, "tags": ["h5", "monoclinic"]},
  {"id": "8ys9", "input": "8ys9/8YS9_xray-data_8ys9/data/SDS_09_18112_master.h5", "wavelength": 0.97957, "ref": {"sg": "P 21 21 21", "sgno": 19, "cell": [71.04, 77.68, 83.16, 90.0, 90.0, 90.0], "dmin": 1.46}, "tags": ["h5", "orthorhombic"]},
  {"id": "9b22", "input": "9b22/KlpnC_20447_a_B1-AR6-AMP_9b22/data/PSL-1011_1580_master.h5", "wavelength": 0.97856, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [39.826, 92.659, 57.716, 90.0, 91.68, 90.0], "dmin": 1.3}, "tags": ["h5", "monoclinic"], "tiers": {"ci": "HDF5, monoclinic P21 at 1.1 A"}},
@@ -186,7 +186,7 @@
  {"id": "2xfw", "input": "2xfw/E192N_alone_MS_1_001.img", "wavelength": 0.9796, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [78.11, 116.53, 83.74, 90, 117.95, 90], "dmin": 1.65}, "tags": ["smv", "monoclinic", "twin", "pseudo-merohedral"]},
  {"id": "2wnn", "input": "2wnn/wt_pyr_pos3_MS_3_001.img", "wavelength": 0.92, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [54.693, 142.456, 83.626, 90, 109.16, 90], "dmin": 1.65}, "tags": ["smv", "monoclinic", "twin", "pseudo-merohedral"]},
  {"id": "2wnz", "input": "2wnz/E192_pyr_MS_1_001.img", "wavelength": 0.9796, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [56.921, 143.041, 83.918, 90, 109.79, 90], "dmin": 1.85}, "tags": ["smv", "monoclinic", "pseudo-symmetry"]},
- {"id": "6oww", "input": "6oww/data/t1_1_00001.cbf", "wavelength": 1.3857, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [85.832, 86.015, 214.34, 90.0, 89.91, 90.0], "dmin": 3.84}, "ref_alternatives": [{"sg": "P 41 21 2", "sgno": 92, "why": "OPEN QUESTION, both accepted. The deposited 8-chain P 1 21 1 model itself obeys a 4-fold (0.13-0.16 A) and a 2-fold (0.10 A) with consistent translations, and the model's |Fc|^2 correlate across the added 422 operators at 1.00/0.98/0.97 - the deposition is likely under-called, and our data support point group 422. Not established: no refinement in P 41 21 2 has been compared with the deposited P 1 21 1, so a pseudo-symmetry too exact for these tests remains a live explanation, as does the deposited assignment (cf. 6toc)"}], "symmetry_evidence": "symmetry/6oww", "tags": ["cbf", "monoclinic", "pseudo-symmetry"]},
+ {"id": "6oww", "input": "6oww/data/t1_1_00001.cbf", "wavelength": 1.3857, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [85.832, 86.015, 214.34, 90.0, 89.91, 90.0], "dmin": 3.84}, "ref_alternatives": [{"sg": "P 41 21 2", "sgno": 92, "why": "OPEN QUESTION, both accepted. The deposited 8-chain P 1 21 1 model obeys a 4-fold (0.13-0.16 A) and a 2-fold (0.10 A), and its |Fc|^2 correlate across the added 422 operators at 1.00/0.98/0.97. POINTLESS reads 4/mmm on our observations (0.98) and on the depositor's amplitudes (0.999). ZANUDA 1.097 with the deposited model prefers P 43 21 2 on our data (three free sets, R-free 0.414-0.430 against 0.435-0.447) and on the depositor's (0.333 against 0.359). A higher symmetry is consistent with our data and the model; the model's hand is P 43 21 2. An exact pseudo-symmetry may explain this (evidence: tools/battery/symmetry/6oww)"}], "symmetry_evidence": "symmetry/6oww", "tags": ["cbf", "monoclinic", "pseudo-symmetry"]},
  {"id": "6p8j", "input": "6p8j/13_12_2_000001.cbf", "wavelength": 0.9792, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [46.484, 152.166, 88.255, 90.0, 90.12, 90.0], "dmin": 1.47}, "symmetry_evidence": "symmetry/6p8j", "tags": ["cbf", "monoclinic", "pseudo-symmetry"]},
  {"id": "9qvv", "input": "9qvv/CHRD_HEP_M/CHRD_404_7_1.nxs", "wavelength": 0.9762, "ref": {"sg": "I 2 2 2", "sgno": 23, "cell": [120.142, 123.8, 123.787, 90.0, 90.0, 90.0], "dmin": 2.72}, "tags": ["nxs", "orthorhombic", "pseudo-symmetry"]},
  {"id": "5ojv", "input": "5ojv/bg3047_w1_1_0001.cbf", "wavelength": 0.97625, "ref": {"sg": "P 21 21 2", "sgno": 18, "cell": [87.792, 158.242, 87.589, 90.0, 90.0, 90.0], "dmin": 2.062}, "tags": ["cbf", "orthorhombic", "pseudo-symmetry"]},
diff --git a/writer/FileWriter.cpp b/writer/FileWriter.cpp
index 9ac107781..5d6b42432 100644
--- a/writer/FileWriter.cpp
+++ b/writer/FileWriter.cpp
@@ -87,9 +87,11 @@ void FileWriter::WriteHDF5(const DataMessage& msg) {
         if (files.size() <= file_number)
             files.resize(file_number + 1);
 
-        if (!files[file_number])
+        if (!files[file_number]) {
             files[file_number] = std::make_unique<HDF5DataFile>(start_message, file_number,
                                                                 HDF5Metadata::DataFileName(start_message, file_number));
+            open_files.insert(file_number);
+        }
         files[file_number]->Write(msg, image_number);
 
         if (files[file_number]->GetNumImages() == start_message.images_per_file) {
@@ -111,6 +113,7 @@ void FileWriter::CloseFile(uint64_t file_number) {
     auto file_stats = files[file_number]->Close();
     files[file_number].reset();
     closed_files.insert(file_number);
+    open_files.erase(file_number);
 
     AddStats(file_stats);
 }
@@ -119,14 +122,15 @@ void FileWriter::CloseOldFiles(uint64_t current_image_number) {
     if (start_message.images_per_file == 0)
         return;
 
-    for (uint64_t f = 0; f < files.size(); ++f) {
-        if (!files[f] || closed_files.contains(f))
-            continue;
-
+    // Ascending, so the first file that is not old enough ends the walk; CloseFile erases from the set,
+    // hence the iterator is advanced first.
+    for (auto it = open_files.begin(); it != open_files.end();) {
+        const uint64_t f = *it++;
         const uint64_t file_end_image = (f + 1) * start_message.images_per_file - 1;
-        if (current_image_number > file_end_image + close_file_lag_images) {
+        if (current_image_number > file_end_image + close_file_lag_images)
             CloseFile(f);
-        }
+        else
+            break;
     }
 }
 
diff --git a/writer/FileWriter.h b/writer/FileWriter.h
index 1bd0965ae..c4ba5494e 100644
--- a/writer/FileWriter.h
+++ b/writer/FileWriter.h
@@ -10,6 +10,7 @@
 #include "../common/ZMQWrappers.h"
 #include "HDF5NXmx.h"
 #include <functional>
+#include <set>
 #include <unordered_set>
 
 class FileWriter {
@@ -23,6 +24,10 @@ class FileWriter {
     std::function<void(const HDF5DataFileStatistics &)> file_closed_callback;
 
     std::unordered_set<uint64_t> closed_files;
+    // File numbers with a live HDF5DataFile, in order. CloseOldFiles walks this and not `files`, whose
+    // slots stay (empty) for every file of the run - a thousand of them for a long run, visited on every
+    // image.
+    std::set<uint64_t> open_files;
     constexpr static uint64_t close_file_lag_images = 1000;
     constexpr static uint64_t default_images_per_file = 1000;
     void CreateHDF5MasterFile(const StartMessage& msg);
diff --git a/writer/StreamWriter.cpp b/writer/StreamWriter.cpp
index af1bde36f..efac1d632 100644
--- a/writer/StreamWriter.cpp
+++ b/writer/StreamWriter.cpp
@@ -178,7 +178,12 @@ void StreamWriter::ProcessDataImage() {
                 if (verbose)
                     logger.Info("Written");
                 NotifyTcpAck(TCPFrameType::DATA, true, false, TCPAckCode::None);
-            } catch (const JFJochException &e) {
+            } catch (const std::exception &e) {
+                // std::exception, not JFJochException, for the reason given in ProcessStartMessage: an
+                // exception that leaves this function skips the error state and the fatal ACK, and the
+                // broker then waits for an END ACK that never comes. Not every thrower below is ours -
+                // std::bad_alloc from a per-image array, filesystem_error from a path check, the JSON
+                // library when a file-done notification carries a string that is not UTF-8.
                 logger.ErrorException(e);
                 logger.Warning("Error writing image - switching to error state");
                 state = StreamWriterState::Error;
@@ -206,7 +211,11 @@ void StreamWriter::ProcessEndMessage() {
             if ((image_puller_output.cbor->end_message->max_image_number == 0) && (max_image_number > 0))
                 image_puller_output.cbor->end_message->max_image_number = max_image_number;
             file_writer->WriteHDF5(*image_puller_output.cbor->end_message);
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
+            // std::exception for the same reason as in ProcessDataImage. Here the cost of a leak is the
+            // highest: the END ACK and the notification below are never sent, the state never reaches
+            // Finalized, and on TCP the broker waits for the ACK for as long as the writer keeps
+            // heartbeating - which it does, whatever state it is in.
             logger.ErrorException(e);
             logger.Error("Error writing end message - switching to error state");
             state = StreamWriterState::Error;
@@ -333,9 +342,9 @@ StreamWriterOutput StreamWriter::Run() {
     StreamWriterOutput ret;
     ret.data_file_stats = hdf5_data_file_statistics;
     ret.stats = GetStatistics();
-    logger.Info("Write task done. Images = {} Throughput = {:.0f} MB/s Frame rate = {:.0f} Hz max occupation of FIFO {}",
+    logger.Info("Write task done. Images = {} Throughput = {:.0f} MB/s Frame rate = {:.0f} Hz max occupation of FIFO {} ring buffer peak {} MB",
                 ret.stats.processed_images, ret.stats.performance_MBs, ret.stats.performance_Hz,
-                ret.stats.max_puller_fifo_utilization);
+                ret.stats.max_puller_fifo_utilization, image_puller.GetMaxRingBufferUtilization() >> 20);
     return ret;
 }
 
@@ -411,7 +420,8 @@ void StreamWriter::NotifyReceiverOnFinalizedWrite(const std::string &detector_up
         s.SendTimeout(std::chrono::seconds(1));
         s.Connect(detector_update_zmq_addr);
         s.Send(j.dump());
-    } catch (const JFJochException &e) {
+    } catch (const std::exception &e) {
+        // std::exception: json::dump throws its own type when run_name is not valid UTF-8.
         logger.ErrorException(e);
         logger.Error("Error sending notification to detector update socket");
     }
diff --git a/writer/jfjoch_writer.cpp b/writer/jfjoch_writer.cpp
index f5c4001f0..c8ee45b92 100644
--- a/writer/jfjoch_writer.cpp
+++ b/writer/jfjoch_writer.cpp
@@ -25,7 +25,8 @@ void print_usage() {
     logger.Info("-d<path>| --root_dir=<path>          Root directory for file writing");
     logger.Info("-r<int> | --zmq_repub_port=<int>     ZeroMQ port for PUSH socket to republish images");
     logger.Info("-f<int> | --zmq_file_port=<int>      ZeroMQ port for PUB socket to inform about finalized files");
-    logger.Info("-w<int> | --rcv_watermark=<int>      Receiving ZeroMQ socket watermark (default = 100)");
+    logger.Info("-w<int> | --rcv_watermark=<int>      Receiving ZeroMQ socket watermark in messages (default = 10)");
+    logger.Info("-m<int> | --rcv_ring_mb=<int>        Receive ring buffer in MB; holds images until written and republished (default = 2048)");
     logger.Info("-W<int> | --repub_watermark=<int>    Republish ZeroMQ socket watermark (default = 1000)");
     logger.Info("-b<int> | --rcv_buffer=<int>         Receiving TCP socket buffer in bytes (only with -T; disables autotuning, capped by net.core.rmem_max)");
     logger.Info("-v      | --verbose                  Verbose output");
@@ -74,6 +75,8 @@ int main(int argc, char **argv) {
     std::optional<int32_t> rcv_watermark;
     std::optional<int32_t> repub_watermark;
     std::optional<int32_t> rcv_buffer;
+    // jfjoch_writer always runs with a ring; the puller's own default (no ring) is for the broker.
+    size_t ring_size = static_cast<size_t>(2048) << 20;
     bool verbose = false;
 
     static struct option long_options[] = {
@@ -84,6 +87,7 @@ int main(int argc, char **argv) {
             {"rcv_watermark", required_argument, 0, 'w'},
             {"repub_watermark", required_argument, 0, 'W'},
             {"rcv_buffer", required_argument, 0, 'b'},
+            {"rcv_ring_mb", required_argument, 0, 'm'},
             {"verbose", no_argument, 0, 'v'},
             {"tcp", no_argument, 0, 'T'},
             {"nproc", required_argument, 0, 'j'},
@@ -93,7 +97,7 @@ int main(int argc, char **argv) {
     int option_index = 0;
     int opt;
     bool raw_tcp = false;
-    while ((opt = getopt_long(argc, argv, "?hH:r:f:R:d:W:w:b:vTj:",long_options, &option_index)) != -1 ) {
+    while ((opt = getopt_long(argc, argv, "?hH:r:f:R:d:W:w:b:m:vTj:",long_options, &option_index)) != -1 ) {
         switch (opt) {
             case 'j':
                 nforks = atoi(optarg);
@@ -129,6 +133,13 @@ int main(int argc, char **argv) {
                     exit(EXIT_FAILURE);
                 }
                 break;
+            case 'm':
+                if (atoi(optarg) <= 0) {
+                    std::cerr << "Ring buffer size must be positive number of MB" << std::endl;
+                    exit(EXIT_FAILURE);
+                }
+                ring_size = static_cast<size_t>(atoi(optarg)) << 20;
+                break;
             case 'W':
                 repub_watermark = atoi(optarg);
                 if (repub_watermark.value() <= 0) {
@@ -240,9 +251,12 @@ int main(int argc, char **argv) {
 
     std::unique_ptr<ImagePuller> puller;
     if (raw_tcp)
-        puller = std::make_unique<TCPImagePuller>(argv[first_argc], rcv_buffer, repub_zmq_addr, repub_watermark);
+        puller = std::make_unique<TCPImagePuller>(argv[first_argc], rcv_buffer, repub_zmq_addr, repub_watermark,
+                                                  ring_size);
     else
-        puller = std::make_unique<ZMQImagePuller>(argv[first_argc], repub_zmq_addr, rcv_watermark, repub_watermark);
+        puller = std::make_unique<ZMQImagePuller>(argv[first_argc], repub_zmq_addr, rcv_watermark, repub_watermark,
+                                                  ring_size);
+    logger.Info("Receive ring buffer {} MB", puller->GetRingBufferCapacity() >> 20);
 
 
     writer = new StreamWriter(logger,*puller,file_done_zmq_addr, verbose);