Cut fire-history write load and register meters lazily

docs/docs/backends/aerospike.md

@@ -124,24 +124,74 @@ was actually read, so a crash leaves it short, which keeps the shard active and
 
 ## Fire history
 
-When `fireHistoryEnabled` is set, each shard's **unified metadata record** carries an extra map bin
-holding delivery-time checkpoints: window start to the fire pointer reached in that window.
-
-- **One entry per shard per window**, regardless of throughput.
-- **Checkpoints are written during the active-shard refresh, not on the claim path.** That refresh
-  already batch-reads every shard's pointer record, so the numbers being recorded are already in
-  hand and no extra read is needed. Deliberately *not* coupled to the claim: claims are the hot path
-  and are far more frequent than the resolution a window needs.
-- `firePointerBefore` also records before it resolves, so a magazine whose history cannot yet answer
-  still accumulates usable history rather than being stuck unable to ever answer.
-- **Each shard's checkpoint is claimed independently**, and the claim is released if the write fails,
-  so a shard that errored is retried next round instead of sitting the window out.
+When `fireHistoryEnabled` is set, each shard's **unified metadata record** carries an extra map bin,
+`FIRE_HISTORY`, holding delivery-time checkpoints: `label -> fire pointer`, meaning "by `label`, the
+fire pointer had reached at least this value". Keys are stored negated so the key-ordered map sorts
+newest first. The logic lives in `FireHistoryStore`.
+
+### Windows and labels
+
+Time is cut into fixed windows of `fireHistoryWindowSeconds`, aligned to the epoch. Each shard gets
+**at most one checkpoint per window**, regardless of throughput or how many processes serve the
+magazine.
+
+A checkpoint is labelled with the **end** of the window it was observed in. The label must never be
+earlier than the moment the pointer was read, or an answer would claim slots were taken before they
+were; the end of the window is always at or after the observation. The consequence is that a checkpoint
+becomes usable one window after it is taken.
+
+### When a checkpoint is written
+
+Checkpoints are written from the **active-shard refresh**, not the claim path. The refresh already
+batch-reads every shard's pointer record; with fire history on, that same read also fetches each
+shard's **newest checkpoint only** (one map entry, not the whole map). A shard is then skipped, with
+no call at all, when:
+
+- **This window is already recorded** — its newest label is this window's. Another process, or an
+  earlier call here, wrote it.
+- **The pointer has not moved** — its newest checkpoint already holds the current fire pointer. The
+  fire pointer only increases, so that older checkpoint stays true for every later instant. An idle
+  magazine costs nothing, however many processes serve it.
+
+Every remaining shard is written in **one batch call per magazine**, not one call per shard.
+
+### Spreading writes across processes
+
+Every consuming process runs the refresh. To stop them all writing right after each window boundary,
+each process waits a **random offset into the first half of the window**, drawn once per magazine and
+kept for the life of the process. The first process past its offset writes; the others read that
+checkpoint on their next refresh and skip. This is always on with fire history and is not
+configurable.
+
+### Writes from `firePointerBefore`
+
+`firePointerBefore` reads the whole history, then applies the same rules before resolving — never
+delayed by the offset. This keeps history advancing when nothing is consuming (the refresh only runs
+while something calls `fire()`), so claims made just before consumption stopped still become visible.
+In steady state the refresh has already recorded the window and this writes nothing.
+
+### Safety and retention
+
+- **Best-effort.** A failed write leaves the previous checkpoint in place, which is still true. It
+  never fails the refresh or the read that triggered it.
+- **Per-shard claims in each process** stop concurrent callers here from writing the same window,
+  and a claim is released if its write fails, so that shard is retried on the next refresh.
+- **Create-only per label.** If two processes race on a label, the first wins and the other is a
+  server-side no-op.
 - **Eviction is by count** (`fireHistoryEntries`), applied on write — by count rather than age so an
   idle magazine keeps reach instead of having its history wiped by the first write after a pause.
+  The map rides the pointer record that every claim rewrites, so larger values make every claim
+  slightly heavier.
 - **Legacy split metadata does not support it.** Fire history requires the unified metadata record.
 
-The cost is bytes on a record that is already read and written by a periodic refresh, not an extra
-operation on the delivery path.
+### Cost
+
+| Magazine state | Calls per window |
+|---|---|
+| Idle (pointer unchanged) | none |
+| Active | usually one batch write for the whole magazine, from one process |
+
+Plus one map entry per shard on a refresh read that already happens.
 
 ## Metadata schema versions
 

docs/docs/concepts/api-reference.md

@@ -37,8 +37,8 @@ Magazine<T> magazine = Magazine.<T>builder()
 | `getShards()` | `int` | This magazine's **persisted** shard count (from the resolved `MagazineContext`), not the shard count configured on the storage. |
 | `peek(Map<Integer, Set<Long>> shardPointersMap)` | `Set<MagazineData<T>>` | Read specific records without consuming. |
 | `getMagazineIdentifier()` | `String` | The identifier this magazine is bound to. |
-| `firePointerBefore(Instant instant)` | `Map<String, FireCheckpoint>` | Where each shard's fire pointer stood at `instant`. Requires `fireHistoryEnabled`. A shard is **absent** when no checkpoint reaches back that far — treat that as "do nothing here", not "empty". Throws `NOT_ENABLED`, or `INVALID_REQUEST` when the retained history no longer spans `instant`. |
-| `fireHistory()` | `Map<String, List<FireCheckpoint>>` | Every retained checkpoint per shard, newest first. Requires `fireHistoryEnabled`. |
+| `firePointerBefore(Instant instant)` | `Map<String, FireCheckpoint>` | Where each shard's fire pointer stood at `instant`. Requires `fireHistoryEnabled`. A shard is **absent** when no checkpoint reaches back that far — treat that as "do nothing here", not "empty". Throws `NOT_ENABLED`, or `INVALID_REQUEST` when the retained history no longer spans `instant`. Also records the current pointer if this window has no checkpoint yet and the pointer moved — usually nothing. |
+| `fireHistory()` | `Map<String, List<FireCheckpoint>>` | Every retained checkpoint per shard, newest first. Requires `fireHistoryEnabled`. Read-only: never records a checkpoint. |
 
 !!! warning "`fire()` is at-most-once"
     A record is claimed by a guarded atomic increment of the fire pointer. If the client times out after the server applied the claim, the pointer has advanced and that record will never be delivered. `fire()` throws `NOTHING_TO_FIRE` when the magazine is drained and `RETRIES_EXHAUSTED` when it gave up while skipping pointer holes — the latter does **not** mean the queue is empty. See [Delivery Semantics](delivery-semantics.md).

docs/docs/concepts/defaults.md

@@ -15,7 +15,7 @@
 | `maxFireHoleSkips` | `512` | Bounds `fire()`; exhausting it raises `RETRIES_EXHAUSTED`. |
 | `metricsEnabled` | `true` | Publishes to Micrometer's global registry unless a `meterRegistry` is set on the builder. |
 | `fireHistoryEnabled` | `false` | Records delivery-time checkpoints, which `firePointerBefore` reads. Off by default: the checkpoint map rides the pointer record. |
-| `fireHistoryWindowSeconds` | `300` | Seconds per checkpoint, and so the resolution of every `firePointerBefore` answer. |
+| `fireHistoryWindowSeconds` | `300` | Seconds per checkpoint, and so the resolution of every `firePointerBefore` answer. Background writes are spread over the first half of each window; not configurable. |
 | `fireHistoryEntries` | `32` | Checkpoints retained per shard. Evicted by **count**, not age. |
 
 ### `metaDataTtl` must outlive `recordTtl`

docs/docs/concepts/metrics.md

@@ -100,7 +100,9 @@ regardless of who attached it.
 
 ## What is published
 
-All meters are tagged `magazine`.
+All meters are tagged `magazine`. Each is registered the first time it records, so a magazine that
+has never fired carries no fire meters, and an operation that never ran has no counter. An absent
+meter reads as zero in rates and sums.
 
 | Meter | Type | Tags | What it tells you |
 |---|---|---|---|
@@ -118,7 +120,11 @@ All meters are tagged `magazine`.
 `claim_fire_pointer`, `increment_load_pointer`, `increment_load_counter`, `increment_fire_counter`,
 `decrement_fire_counter`, `batch_read_metadata`, `refresh_active_shards`, `read_data`, `write_data`,
 `delete_data`, `batch_delete_data`, `batch_read_data`, `claim_dedupe_marker`,
-`withdraw_dedupe_marker`.
+`withdraw_dedupe_marker`, `write_fire_history`, `read_fire_history`.
+
+`write_fire_history` counts **batch calls**: at most one per magazine per window, and none for a
+magazine whose fire pointers have not moved. `read_fire_history` is the read behind
+`firePointerBefore` and `fireHistory`.
 
 `refresh_active_shards` is the batch read behind an active-shard cache refresh, split out from
 `batch_read_metadata` so refresh traffic is attributable without a meter of its own. A magazine on
@@ -138,6 +144,8 @@ Divide `magazine.aerospike.calls` by the matching outcome counter. Healthy stead
 | Shard discovery | 1–2 batches / `activeShardRefreshSeconds` / magazine | `refresh_active_shards`; fans out to every shard |
 | `delete()` | **1** | `delete_data` |
 | `deleteAll()` | **1** | `batch_delete_data`, whatever the batch size. A one-record batch takes the single-key path and counts as `delete_data` |
+| Fire-history recording | **≤ 1** batch / window / magazine | `write_fire_history`, only if some shard's fire pointer moved and the window is not yet recorded by any process. Rides the shard-discovery read; no extra read |
+| `firePointerBefore()` | **1**, plus **≤ 1** | `read_fire_history`, plus a `write_fire_history` batch only when the read finds the window unrecorded |
 
 **Calls per `fire()` does not rise with consumer count.** The claim is a guarded atomic increment,
 so concurrent consumers get distinct pointers and no claim is ever lost. If you see the ratio climb

docs/docs/upgrading.md

@@ -125,6 +125,23 @@ classpath. Declare it directly if you depended on it transitively.
 - **`Magazine.deleteAll(Collection<MagazineData<T>>)`** — retire a whole batch in one round trip.
   Optional for callers; **mandatory for anyone implementing `BaseMagazineStorage`**, see section 4.
 
+## Upgrading within 2.0
+
+### 2.0.2
+
+No API changes. Behaviour that can show up on dashboards and alerts:
+
+- **Fire history writes far less.** Checkpoints are written in one batch per magazine, only for shards
+  whose fire pointer moved, and normally by one process per window. `write_fire_history` now counts
+  batch calls, so expect a large drop. Batch writes need Aerospike server 6.0 or later, which batch
+  deletes already required.
+- **Checkpoints are labelled with the end of their window**, not the start, so a label is never
+  earlier than the observation. `firePointerBefore` therefore answers about one window later than
+  before. Entries written by 2.0.1 keep their window-start labels and remain readable; a mixed
+  deployment during a rolling upgrade is safe.
+- **Meters are registered on first use.** An operation or outcome that never happened has no meter;
+  rates and sums read it as zero, but an alert that expects the series to exist needs a default.
+
 ## Checklist
 
 1. Change the artifact to `magazine-core`.

docs/docs/usage.md

@@ -157,6 +157,10 @@ message.
 **Off by default.** The checkpoint map rides the shard's pointer record and every claim rewrites that
 record, so a caller that never asks should not pay for it.
 
+Once on, checkpoints are recorded automatically while the magazine is consumed: at most one per shard
+per window, none for a shard whose pointer has not moved, and one batch call per magazine. See
+[Fire history](backends/aerospike.md#fire-history) for how.
+
 ```java
 AerospikeStorageConfig.builder()
         .dataSetName("my_data")
@@ -183,6 +187,9 @@ watermarks.forEach((shard, checkpoint) ->
 Every slot at or below the returned pointer was claimed at or before that instant. Slots above it
 might have been claimed a moment ago.
 
+A checkpoint is labelled with the end of the window it was observed in, so it answers questions about
+instants from one window later onwards. Precision is therefore one to two windows.
+
 !!! note "An absent shard means 'do nothing', not 'empty'"
     A shard is missing from the result when no checkpoint reaches back that far yet. That is normal
     for a young magazine. Treat it as "nothing is provably old enough here this round" — never as
@@ -192,8 +199,8 @@ might have been claimed a moment ago.
 
 | Setting | Meaning |
 |---|---|
-| `fireHistoryWindowSeconds` | Seconds covered by one checkpoint, and therefore the **resolution** of every answer. One entry per shard per window, whatever the throughput |
-| `fireHistoryEntries` | Checkpoints retained per shard. Eviction is by **count, not age**, so an intermittently used magazine reaches further back than `window x entries` rather than having its history wiped by the first write after an idle period |
+| `fireHistoryWindowSeconds` | Seconds covered by one checkpoint, and therefore the **resolution** of every answer. At most one entry per shard per window, whatever the throughput, and none while the shard's pointer does not move. A wider window means fewer writes |
+| `fireHistoryEntries` | Checkpoints retained per shard. Eviction is by **count, not age**, so an intermittently used magazine reaches further back than `window x entries` rather than having its history wiped by the first write after an idle period. The map rides the pointer record every claim rewrites, so raise this only as far as you need |
 
 Asking about an instant further back than the retained history throws `INVALID_REQUEST`, and asking
 at all with the feature disabled throws `NOT_ENABLED`.

magazine-core/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>com.phonepe</groupId>
         <artifactId>magazine</artifactId>
-        <version>2.0.1-SNAPSHOT</version>
+        <version>2.0.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>magazine-core</artifactId>

magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/AerospikeStorage.java

@@ -34,6 +34,7 @@
 import com.phonepe.magazine.impl.aerospike.common.ErrorMessage;
 import com.phonepe.magazine.impl.aerospike.store.ActiveShardSelector;
 import com.phonepe.magazine.impl.aerospike.store.DeDupeGuard;
+import com.phonepe.magazine.impl.aerospike.store.FireHistoryStore;
 import com.phonepe.magazine.impl.aerospike.store.MagazineDataStore;
 import com.phonepe.magazine.impl.aerospike.store.MagazineMetadataStore;
 import com.phonepe.magazine.metrics.MagazineMetrics;
@@ -106,7 +107,7 @@ public AerospikeStorage(
                 namespace, metaSetName, storageConfig.getShards(), storageConfig.isAllowShardIncrease());
         this.metadataStore = new MagazineMetadataStore(aerospikeClient, retryerFactory, metrics,
                 namespace, metaSetName, getMetaDataTtl(),
-                new MagazineMetadataStore.FireHistoryConfig(storageConfig.isFireHistoryEnabled(),
+                new FireHistoryStore.FireHistoryConfig(storageConfig.isFireHistoryEnabled(),
                         storageConfig.fireHistoryWindowMillis(), storageConfig.getFireHistoryEntries()));
         this.dataStore = new MagazineDataStore<>(aerospikeClient, retryerFactory, metrics,
                 namespace, dataSetName, getRecordTtl(), clazz);

magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/common/AerospikeConstants.java

@@ -42,9 +42,8 @@ public final class AerospikeConstants {
     public static final String CREATED_AT = "CREATED_AT";
     public static final String SHARDS_BIN = "SHARDS";
     /**
-     * Delivery-time checkpoints: a key-ordered map of {@code negated window start -> fire pointer}.
-     * Keys are negated so the map sorts newest-first, which makes eviction a single index-range
-     * removal that keeps the newest N entries.
+     * Delivery-time checkpoints: key-ordered map of {@code -label -> fire pointer}, label being the
+     * observation window's end. Negated keys sort newest-first, so eviction is one index range.
      */
     public static final String FIRE_HISTORY = "FIRE_HISTORY";
 

magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/common/AerospikePolicies.java

@@ -19,6 +19,7 @@
 import com.aerospike.client.IAerospikeClient;
 import com.aerospike.client.policy.BatchDeletePolicy;
 import com.aerospike.client.policy.BatchPolicy;
+import com.aerospike.client.policy.BatchWritePolicy;
 import com.aerospike.client.policy.WritePolicy;
 import java.util.Objects;
 import lombok.AccessLevel;
@@ -52,4 +53,10 @@ public static BatchDeletePolicy batchDeletePolicy(final IAerospikeClient client)
         final BatchDeletePolicy defaults = client.getBatchDeletePolicyDefault();
         return Objects.isNull(defaults) ? new BatchDeletePolicy() : new BatchDeletePolicy(defaults);
     }
+
+    /** @return a copy of the client's default batch write policy, or a fresh default. */
+    public static BatchWritePolicy batchWritePolicy(final IAerospikeClient client) {
+        final BatchWritePolicy defaults = client.getBatchWritePolicyDefault();
+        return Objects.isNull(defaults) ? new BatchWritePolicy() : new BatchWritePolicy(defaults);
+    }
 }