diff --git a/docs/docs/backends/aerospike.md b/docs/docs/backends/aerospike.md index 0e5fdb6..8438e01 100644 --- a/docs/docs/backends/aerospike.md +++ b/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 diff --git a/docs/docs/concepts/api-reference.md b/docs/docs/concepts/api-reference.md index f7b6aa7..33629da 100644 --- a/docs/docs/concepts/api-reference.md +++ b/docs/docs/concepts/api-reference.md @@ -37,8 +37,8 @@ Magazine magazine = Magazine.builder() | `getShards()` | `int` | This magazine's **persisted** shard count (from the resolved `MagazineContext`), not the shard count configured on the storage. | | `peek(Map> shardPointersMap)` | `Set>` | Read specific records without consuming. | | `getMagazineIdentifier()` | `String` | The identifier this magazine is bound to. | -| `firePointerBefore(Instant instant)` | `Map` | 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>` | Every retained checkpoint per shard, newest first. Requires `fireHistoryEnabled`. | +| `firePointerBefore(Instant instant)` | `Map` | 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>` | 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). diff --git a/docs/docs/concepts/defaults.md b/docs/docs/concepts/defaults.md index a0a7c8e..0bc1484 100644 --- a/docs/docs/concepts/defaults.md +++ b/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` diff --git a/docs/docs/concepts/metrics.md b/docs/docs/concepts/metrics.md index 2e16be2..4d567ae 100644 --- a/docs/docs/concepts/metrics.md +++ b/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 diff --git a/docs/docs/upgrading.md b/docs/docs/upgrading.md index 2740a57..c294c73 100644 --- a/docs/docs/upgrading.md +++ b/docs/docs/upgrading.md @@ -125,6 +125,23 @@ classpath. Declare it directly if you depended on it transitively. - **`Magazine.deleteAll(Collection>)`** — 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`. diff --git a/docs/docs/usage.md b/docs/docs/usage.md index 0aa3030..e73b195 100644 --- a/docs/docs/usage.md +++ b/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`. diff --git a/magazine-core/pom.xml b/magazine-core/pom.xml index 2e93619..fb4c23e 100644 --- a/magazine-core/pom.xml +++ b/magazine-core/pom.xml @@ -7,7 +7,7 @@ com.phonepe magazine - 2.0.1-SNAPSHOT + 2.0.2-SNAPSHOT magazine-core diff --git a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/AerospikeStorage.java b/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/AerospikeStorage.java index 9e444df..31d490b 100644 --- a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/AerospikeStorage.java +++ b/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); diff --git a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/common/AerospikeConstants.java b/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/common/AerospikeConstants.java index b195abe..7091013 100644 --- a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/common/AerospikeConstants.java +++ b/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"; diff --git a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/common/AerospikePolicies.java b/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/common/AerospikePolicies.java index 40c460a..97b7a6e 100644 --- a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/common/AerospikePolicies.java +++ b/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); + } } diff --git a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/store/FireHistoryStore.java b/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/store/FireHistoryStore.java new file mode 100644 index 0000000..b9bc4fd --- /dev/null +++ b/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/store/FireHistoryStore.java @@ -0,0 +1,402 @@ +/** + * Copyright (c) 2025 Original Author(s), PhonePe India Pvt. Ltd. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.phonepe.magazine.impl.aerospike.store; + +import com.aerospike.client.BatchRecord; +import com.aerospike.client.BatchWrite; +import com.aerospike.client.IAerospikeClient; +import com.aerospike.client.Key; +import com.aerospike.client.Operation; +import com.aerospike.client.Record; +import com.aerospike.client.ResultCode; +import com.aerospike.client.Value; +import com.aerospike.client.cdt.MapOperation; +import com.aerospike.client.cdt.MapOrder; +import com.aerospike.client.cdt.MapPolicy; +import com.aerospike.client.cdt.MapReturnType; +import com.aerospike.client.cdt.MapWriteFlags; +import com.aerospike.client.policy.BatchPolicy; +import com.aerospike.client.policy.BatchWritePolicy; +import com.aerospike.client.policy.RecordExistsAction; +import com.phonepe.magazine.entity.FireCheckpoint; +import com.phonepe.magazine.entity.MagazineContext; +import com.phonepe.magazine.exception.ErrorCode; +import com.phonepe.magazine.exception.MagazineExceptions; +import com.phonepe.magazine.impl.aerospike.common.AerospikeConstants; +import com.phonepe.magazine.impl.aerospike.common.AerospikePolicies; +import com.phonepe.magazine.impl.aerospike.common.AerospikeRetryer; +import com.phonepe.magazine.impl.aerospike.common.ErrorMessage; +import com.phonepe.magazine.metrics.MagazineMetrics; +import com.phonepe.magazine.metrics.StorageOperation; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.stream.Stream; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; + +/** + * Delivery-time checkpoints: per-shard map of {@code -label -> fire pointer} (newest first) on the + * pointer record. Recorded from the active-shard refresh and from {@code firePointerBefore}; at most + * one per shard per window, labelled with the window end (never earlier than the observation). + * Skipped when the window is already recorded or the pointer is unchanged. One batch write per + * magazine; refresh writes wait for a random per-process offset in the first half of the window. + * Best-effort: failures never fail the triggering call. + */ +@Slf4j +public final class FireHistoryStore { + + /** A second writer for the same label is a server-side no-op. */ + private static final MapPolicy CHECKPOINT_MAP_POLICY = new MapPolicy( + MapOrder.KEY_ORDERED, MapWriteFlags.CREATE_ONLY | MapWriteFlags.NO_FAIL); + + private static final Operation READ_NEWEST_CHECKPOINT = MapOperation.getByIndex( + AerospikeConstants.FIRE_HISTORY, 0, MapReturnType.KEY_VALUE); + + private final IAerospikeClient client; + private final AerospikeRetryer retryer; + private final MagazineMetrics metrics; + private final Function> pointerKeys; + + private final boolean enabled; + private final long windowMillis; + private final int entries; + + private final BatchPolicy batchPolicy; + private final BatchWritePolicy writePolicy; + + /** Bounded by the number of configured magazines. */ + private final Map states = new ConcurrentHashMap<>(); + + /** + * @param windowMillis recording window width; the resolution of every answer. + * @param entries checkpoints retained per shard, evicted by count. + */ + public record FireHistoryConfig(boolean enabled, long windowMillis, int entries) { + } + + public FireHistoryStore(final IAerospikeClient client, + final AerospikeRetryer retryer, + final MagazineMetrics metrics, + final Function> pointerKeys, + final int metaDataTtl, + final FireHistoryConfig config) { + this.client = client; + this.retryer = retryer; + this.metrics = metrics; + this.pointerKeys = pointerKeys; + this.enabled = config.enabled(); + this.windowMillis = config.windowMillis(); + this.entries = config.entries(); + + this.batchPolicy = AerospikePolicies.batchPolicy(client); + // UPDATE_ONLY: never create a pointer record for an uninitialised shard. + this.writePolicy = AerospikePolicies.batchWritePolicy(client); + this.writePolicy.recordExistsAction = RecordExistsAction.UPDATE_ONLY; + this.writePolicy.expiration = metaDataTtl; + } + + /** Extra refresh-read operation: only the newest checkpoint, as {@link #recordOnRefresh} needs. */ + public Optional refreshReadOperation() { + return enabled ? Optional.of(READ_NEWEST_CHECKPOINT) : Optional.empty(); + } + + /** Records from the refresh read (must include the newest checkpoint), after this process's offset. */ + public void recordOnRefresh(final MagazineContext context, final Record[] pointerRecords) { + if (!enabled) { + return; + } + final long now = System.currentTimeMillis(); + final Window window = Window.containing(now, windowMillis); + if (now >= window.start() + state(context).getJitterMillis()) { + recordCheckpoints(context, pointerRecords, window); + } + } + + /** + * Lower bound of each shard's fire pointer at {@code instant}; may lag, never leads. + * + * @return shard id to checkpoint; absent when no checkpoint is old enough yet. + * @throws com.phonepe.magazine.exception.MagazineException {@code NOT_ENABLED} if disabled, + * {@code INVALID_REQUEST} if retained history does not reach {@code instant}. + */ + public Map firePointerBefore(final MagazineContext context, final Instant instant) { + requireEnabled(context); + final Record[] records = readHistory(context); + + // Record first (undelayed) so history still advances even if resolving throws. + recordCheckpoints(context, records, Window.containing(System.currentTimeMillis(), windowMillis)); + + final Map answers = new HashMap<>(capacityFor(context.getShards())); + for (int shard = 0; shard < context.getShards(); shard++) { + final int current = shard; + resolve(context, current, checkpoints(historyBin(records[current])), instant) + .ifPresent(checkpoint -> answers.put(context.shardId(current), checkpoint.toEntity())); + } + return answers; + } + + /** Every retained checkpoint per shard, newest first. Reading never records. */ + public Map> history(final MagazineContext context) { + requireEnabled(context); + final Record[] records = readHistory(context); + final Map> history = new HashMap<>(capacityFor(context.getShards())); + for (int shard = 0; shard < context.getShards(); shard++) { + final List checkpoints = checkpoints(historyBin(records[shard])).stream() + .map(Checkpoint::toEntity) + .toList(); + if (!checkpoints.isEmpty()) { + history.put(context.shardId(shard), checkpoints); + } + } + return history; + } + + // --- recording --------------------------------------------------------------------------- + + private void recordCheckpoints(final MagazineContext context, final Record[] pointerRecords, + final Window window) { + final MagazineState state = state(context); + final List pending = pendingCheckpoints(context, pointerRecords, window, state); + if (!pending.isEmpty()) { + write(context, pending, window, state); + } + } + + /** Shards that need this window's checkpoint, each already claimed for this process. */ + private static List pendingCheckpoints(final MagazineContext context, + final Record[] pointerRecords, + final Window window, + final MagazineState state) { + final List pending = new ArrayList<>(); + for (int shard = 0; shard < context.getShards(); shard++) { + final Record pointers = pointerRecords[shard]; + if (Objects.nonNull(pointers) + && needsCheckpoint(pointers, window) + && state.claim(shard, window)) { + pending.add(new PendingCheckpoint(shard, firePointer(pointers))); + } + } + return pending; + } + + private static boolean needsCheckpoint(final Record pointers, final Window window) { + final Optional newest = checkpoints(pointers.getValue(AerospikeConstants.FIRE_HISTORY)) + .stream().findFirst(); + if (newest.isEmpty()) { + return true; + } + final boolean windowAlreadyRecorded = newest.get().label() >= window.label(); + // Fire pointer is monotonic: an unchanged value stays true, so skip idle shards. + final boolean pointerUnchanged = newest.get().firePointer() == firePointer(pointers); + return !windowAlreadyRecorded && !pointerUnchanged; + } + + /** One batch call for every shard; a failed shard's claim is released so it is retried. */ + private void write(final MagazineContext context, + final List pending, + final Window window, + final MagazineState state) { + final List keys = pointerKeys.apply(context); + final List writes = pending.stream() + .map(checkpoint -> checkpointWrite(keys.get(checkpoint.shard()), window.label(), + checkpoint.firePointer())) + .toList(); + + try { + metrics.aerospikeCall(context.getMagazineIdentifier(), StorageOperation.WRITE_FIRE_HISTORY); + client.operate(batchPolicy, writes); + } catch (Exception e) { + pending.forEach(checkpoint -> state.release(checkpoint.shard(), window)); + log.warn("Could not record fire history for magazine {} window {}.", + context.getMagazineIdentifier(), window.start(), e); + return; + } + + for (int i = 0; i < writes.size(); i++) { + final int resultCode = writes.get(i).resultCode; + if (resultCode == ResultCode.OK) { + continue; + } + final int shard = pending.get(i).shard(); + state.release(shard, window); + // A missing record is expected: the shard has none yet, or it expired. + if (resultCode != ResultCode.KEY_NOT_FOUND_ERROR) { + log.warn("Could not record fire history for magazine {} shard {} window {}: {}.", + context.getMagazineIdentifier(), shard, window.start(), + ResultCode.getResultString(resultCode)); + } + } + } + + private BatchWrite checkpointWrite(final Key key, final long label, final long firePointer) { + return new BatchWrite(writePolicy, key, new Operation[]{ + MapOperation.put(CHECKPOINT_MAP_POLICY, AerospikeConstants.FIRE_HISTORY, + Value.get(Checkpoint.encode(label)), Value.get(firePointer)), + // Evict by count, not age, so a long idle period does not wipe usable history. + MapOperation.removeByIndexRange(AerospikeConstants.FIRE_HISTORY, entries, MapReturnType.NONE)}); + } + + private MagazineState state(final MagazineContext context) { + return states.computeIfAbsent(context, this::newState); + } + + private MagazineState newState(final MagazineContext context) { + // Random offset in the first half of the window, fixed per magazine per process. + final long span = windowMillis / 2; + final long jitter = span <= 0 ? 0L : ThreadLocalRandom.current().nextLong(span); + return new MagazineState(context.getShards(), jitter); + } + + // --- reading ----------------------------------------------------------------------------- + + private Record[] readHistory(final MagazineContext context) { + return retryer.call(() -> { + metrics.aerospikeCall(context.getMagazineIdentifier(), StorageOperation.READ_FIRE_HISTORY); + return client.get(client.getBatchPolicyDefault(), + pointerKeys.apply(context).toArray(new Key[0]), + AerospikeConstants.getFireHistoryBins()); + }); + } + + /** The newest checkpoint labelled at or before {@code instant}. */ + private Optional resolve(final MagazineContext context, + final int shard, + final List checkpoints, + final Instant instant) { + final long cutoff = instant.toEpochMilli(); + final Optional answer = checkpoints.stream() + .filter(checkpoint -> checkpoint.label() <= cutoff) + .findFirst(); + // At capacity with nothing old enough: the answer was evicted; fail rather than return empty. + if (answer.isEmpty() && checkpoints.size() >= entries) { + throw MagazineExceptions.invalidRequest(String.format(ErrorMessage.FIRE_HISTORY_TOO_SHORT, + context.getMagazineIdentifier(), shard, instant, + Instant.ofEpochMilli(checkpoints.get(checkpoints.size() - 1).label()))); + } + return answer; + } + + /** The history bin of a batch-read slot, which is null when the shard has no record. */ + private static Object historyBin(final Record pointerRecord) { + return Objects.isNull(pointerRecord) ? null : pointerRecord.getValue(AerospikeConstants.FIRE_HISTORY); + } + + /** Checkpoints newest first, from either the full map or the refresh read's single entry. */ + private static List checkpoints(final Object raw) { + final Stream rawEntries; + if (raw instanceof Map map) { + rawEntries = map.entrySet().stream(); + } else if (raw instanceof List list) { + rawEntries = list.stream(); + } else { + rawEntries = Stream.ofNullable(raw); + } + return rawEntries.map(Checkpoint::decode) + .flatMap(Optional::stream) + .sorted(Checkpoint.NEWEST_FIRST) + .toList(); + } + + private void requireEnabled(final MagazineContext context) { + if (!enabled) { + throw MagazineExceptions.of(ErrorCode.NOT_ENABLED, String.format( + ErrorMessage.FIRE_HISTORY_DISABLED, context.getMagazineIdentifier())); + } + } + + private static long firePointer(final Record pointers) { + return pointers.getLong(AerospikeConstants.FIRE_POINTER); + } + + private static int capacityFor(final int expectedSize) { + return (int) Math.ceil(expectedSize / 0.75d) + 1; + } + + // --- value types ------------------------------------------------------------------------- + + /** A recording window; {@code label} is its end. */ + private record Window(long start, long label) { + + static Window containing(final long now, final long width) { + final long start = now / width * width; + return new Window(start, start + width); + } + } + + /** A shard to checkpoint, with the fire pointer observed on it. */ + private record PendingCheckpoint(int shard, long firePointer) { + } + + /** By {@code label}, the fire pointer had reached at least {@code firePointer}. */ + private record Checkpoint(long label, long firePointer) { + + static final Comparator NEWEST_FIRST = Comparator.comparingLong(Checkpoint::label).reversed(); + + /** Map keys are negated labels, so the key-ordered map sorts newest-first. */ + static long encode(final long label) { + return -label; + } + + static Optional decode(final Object raw) { + if (raw instanceof Map.Entry entry + && entry.getKey() instanceof Number key + && entry.getValue() instanceof Number value) { + return Optional.of(new Checkpoint(-key.longValue(), value.longValue())); + } + return Optional.empty(); + } + + FireCheckpoint toEntity() { + return new FireCheckpoint(firePointer, Instant.ofEpochMilli(label)); + } + } + + /** Per-process jitter and last claimed window per shard (released on failure for retry). */ + private static final class MagazineState { + private final AtomicLong[] claimedWindows; + @Getter + private final long jitterMillis; + + MagazineState(final int shards, final long jitterMillis) { + this.claimedWindows = new AtomicLong[shards]; + for (int shard = 0; shard < shards; shard++) { + claimedWindows[shard] = new AtomicLong(Long.MIN_VALUE); + } + this.jitterMillis = jitterMillis; + } + + boolean claim(final int shard, final Window window) { + final AtomicLong claimed = claimedWindows[shard]; + final long previous = claimed.get(); + return previous < window.start() && claimed.compareAndSet(previous, window.start()); + } + + void release(final int shard, final Window window) { + claimedWindows[shard].compareAndSet(window.start(), Long.MIN_VALUE); + } + } +} diff --git a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/store/MagazineMetadataStore.java b/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/store/MagazineMetadataStore.java index 3d63943..68e5b2d 100644 --- a/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/store/MagazineMetadataStore.java +++ b/magazine-core/src/main/java/com/phonepe/magazine/impl/aerospike/store/MagazineMetadataStore.java @@ -23,19 +23,13 @@ import com.aerospike.client.Operation; import com.aerospike.client.Record; import com.aerospike.client.ResultCode; -import com.aerospike.client.Value; -import com.aerospike.client.cdt.MapOperation; -import com.aerospike.client.cdt.MapOrder; -import com.aerospike.client.cdt.MapPolicy; -import com.aerospike.client.cdt.MapReturnType; -import com.aerospike.client.cdt.MapWriteFlags; import com.aerospike.client.exp.Exp; +import com.aerospike.client.exp.Expression; import com.aerospike.client.policy.RecordExistsAction; import com.aerospike.client.policy.WritePolicy; import com.phonepe.magazine.entity.FireCheckpoint; import com.phonepe.magazine.entity.MagazineContext; import com.phonepe.magazine.entity.MetaData; -import com.phonepe.magazine.exception.ErrorCode; import com.phonepe.magazine.exception.MagazineExceptions; import com.phonepe.magazine.impl.aerospike.common.AerospikeConstants; import com.phonepe.magazine.impl.aerospike.common.AerospikeNaming; @@ -50,13 +44,9 @@ import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.Optional; import java.util.OptionalLong; -import java.util.SortedMap; -import java.util.TreeMap; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicLong; -import lombok.extern.slf4j.Slf4j; +import java.util.stream.Stream; /** * Reads and writes the per-shard metadata records: load/fire pointers and load/fire counters. @@ -64,27 +54,21 @@ * Pointers and counters live in one record under the unified schema and in two under the legacy * schema; that difference is confined to {@link AerospikeNaming}, so every method here is written * once and works for both. - * + *

+ * Delivery-time checkpoints on the pointer record are owned by {@link FireHistoryStore}. + *

* The filter expression, the write policies and the metadata keys are all constant per magazine * and all resolved once: {@link Exp#build} serialises to a byte array, a policy is a ~25-field * copy, and {@link Key} construction runs a RIPEMD-160 digest. Rebuilding those per operation was * pure overhead on every round trip. */ -@Slf4j public final class MagazineMetadataStore { - /** - * {@code CREATE_ONLY | NO_FAIL} makes a second writer in the same window a server-side no-op - * rather than an error, which is what lets every process attempt the write independently. - */ - private static final MapPolicy CHECKPOINT_MAP_POLICY = new MapPolicy( - MapOrder.KEY_ORDERED, MapWriteFlags.CREATE_ONLY | MapWriteFlags.NO_FAIL); - /** * The claim's precondition: this shard still has an unconsumed slot. Absent bins read as zero, * so a record never fired from does not make the comparison unknown. */ - private static final com.aerospike.client.exp.Expression FIRE_POINTER_BEHIND_LOAD_POINTER = + private static final Expression FIRE_POINTER_BEHIND_LOAD_POINTER = Exp.build(Exp.lt( binOrZero(AerospikeConstants.FIRE_POINTER), binOrZero(AerospikeConstants.LOAD_POINTER))); @@ -97,26 +81,11 @@ public final class MagazineMetadataStore { private final WritePolicy claimPolicy; private final WritePolicy adjustPolicy; - private final WritePolicy checkpointPolicy; - private final boolean fireHistoryEnabled; - private final long fireHistoryWindowMillis; - private final int fireHistoryEntries; + /** Metadata bins, plus the newest checkpoint when fire history is on. */ + private final Operation[] refreshOperations; + private final FireHistoryStore fireHistory; /** Bounded by the number of configured magazines. */ private final Map keyCache = new ConcurrentHashMap<>(); - /** - * Last checkpoint window written, per shard, so only one caller per window issues each write. - * Per shard rather than per magazine: a shard skipped because its record did not exist yet, or - * because its write failed, must be retried within the same window rather than sit out until - * the next one. - *

- * Keyed by context like {@link #keyCache}, because one store serves every magazine on a - * storage. Shard count is part of the context's equality, so the array can never outlive the - * shard count it was sized for. - */ - private final Map checkpointWindows = new ConcurrentHashMap<>(); - - public record FireHistoryConfig(boolean enabled, long windowMillis, int entries) { - } public MagazineMetadataStore(final IAerospikeClient client, final AerospikeRetryer retryerFactory, @@ -124,15 +93,15 @@ public MagazineMetadataStore(final IAerospikeClient client, final String namespace, final String metaSetName, final int metaDataTtl, - final FireHistoryConfig fireHistoryConfig) { + final FireHistoryStore.FireHistoryConfig fireHistoryConfig) { this.client = client; this.retryerFactory = retryerFactory; this.metrics = metrics; this.namespace = namespace; this.metaSetName = metaSetName; - this.fireHistoryEnabled = fireHistoryConfig.enabled(); - this.fireHistoryWindowMillis = fireHistoryConfig.windowMillis(); - this.fireHistoryEntries = fireHistoryConfig.entries(); + this.fireHistory = new FireHistoryStore(client, retryerFactory, metrics, + context -> keys(context).pointerKeys(), metaDataTtl, fireHistoryConfig); + this.refreshOperations = refreshOperations(fireHistory); this.claimPolicy = AerospikePolicies.writePolicy(client); this.claimPolicy.recordExistsAction = RecordExistsAction.UPDATE_ONLY; @@ -144,12 +113,6 @@ public MagazineMetadataStore(final IAerospikeClient client, this.adjustPolicy = AerospikePolicies.writePolicy(client); this.adjustPolicy.recordExistsAction = RecordExistsAction.UPDATE; this.adjustPolicy.expiration = metaDataTtl; - - // UPDATE_ONLY: a checkpoint describes a shard that exists. Creating the record here would - // fabricate a pointer record for a shard that was never initialised. - this.checkpointPolicy = AerospikePolicies.writePolicy(client); - this.checkpointPolicy.recordExistsAction = RecordExistsAction.UPDATE_ONLY; - this.checkpointPolicy.expiration = metaDataTtl; } /** @@ -259,185 +222,32 @@ public Map readAll(final MagazineContext context) { * @return shard numbers with data available to fire, ascending. */ public int[] activeShards(final MagazineContext context) { - final Record[] pointerRecords = batchRead(context, keys(context).pointerKeys(), - StorageOperation.REFRESH_ACTIVE_SHARDS); + final Record[] pointerRecords = refreshRead(context); final Record[] counterRecords = AerospikeNaming.usesUnifiedMetadata(context) ? pointerRecords : batchRead(context, keys(context).counterKeys(), StorageOperation.REFRESH_ACTIVE_SHARDS); - // The batch read above already holds every shard's fire pointer, so recording checkpoints - // here costs no additional read, and this cache refreshes asynchronously. - recordFireHistory(context, pointerRecords); + // Reuses the read above; no extra call. + fireHistory.recordOnRefresh(context, pointerRecords); final int[] active = new int[context.getShards()]; int found = 0; for (int shard = 0; shard < context.getShards(); shard++) { - final Record pointers = pointerRecords[shard]; - final Record counters = counterRecords[shard]; - if (Objects.nonNull(pointers) - && Objects.nonNull(counters) - && counters.getLong(AerospikeConstants.LOAD_COUNTER) > counters.getLong(AerospikeConstants.FIRE_COUNTER) - && pointers.getLong(AerospikeConstants.LOAD_POINTER) > pointers.getLong(AerospikeConstants.FIRE_POINTER)) { + if (isActive(pointerRecords[shard], counterRecords[shard])) { active[found++] = shard; } } return found == active.length ? active : Arrays.copyOf(active, found); } - /** - * Where each shard's fire pointer stood at {@code instant}. - *

- * Answers come from checkpoints recorded at or before {@code instant}, so a returned pointer is - * a lower bound: every slot at or below it was claimed by then. It may lag reality, never lead - * it. - * - * @return shard id to checkpoint. A shard is absent when no checkpoint is old enough yet. - * @throws com.phonepe.magazine.exception.MagazineException {@code NOT_ENABLED} when fire - * history is disabled, {@code INVALID_REQUEST} when retained history does not reach - * back to {@code instant}. - */ + /** @see FireHistoryStore#firePointerBefore */ public Map firePointerBefore(final MagazineContext context, final Instant instant) { - if (!fireHistoryEnabled) { - throw MagazineExceptions.of(ErrorCode.NOT_ENABLED, String.format( - ErrorMessage.FIRE_HISTORY_DISABLED, context.getMagazineIdentifier())); - } - final Record[] records = readFireHistory(context); - - // Recorded before resolving, so a shard whose history cannot answer still advances: the - // throw below must not stop this magazine from ever accumulating usable history again. - recordFireHistory(context, records); - - final Map checkpoints = new HashMap<>(capacityFor(context.getShards())); - for (int shard = 0; shard < context.getShards(); shard++) { - final int current = shard; - resolveCheckpoint(context, records[current], instant, current) - .ifPresent(checkpoint -> checkpoints.put(context.shardId(current), checkpoint)); - } - return checkpoints; + return fireHistory.firePointerBefore(context, instant); } - /** - * Every retained checkpoint, newest first - */ + /** @see FireHistoryStore#history */ public Map> fireHistory(final MagazineContext context) { - if (!fireHistoryEnabled) { - throw MagazineExceptions.of(ErrorCode.NOT_ENABLED, String.format( - ErrorMessage.FIRE_HISTORY_DISABLED, context.getMagazineIdentifier())); - } - final Record[] records = readFireHistory(context); - final Map> history = new HashMap<>(capacityFor(context.getShards())); - for (int shard = 0; shard < context.getShards(); shard++) { - final List checkpoints = readHistory(records[shard]).entrySet().stream() - .map(entry -> new FireCheckpoint(entry.getValue(), Instant.ofEpochMilli(-entry.getKey()))) - .toList(); - if (!checkpoints.isEmpty()) { - history.put(context.shardId(shard), checkpoints); - } - } - return history; - } - - private Record[] readFireHistory(final MagazineContext context) { - return retryerFactory.call(() -> { - metrics.aerospikeCall(context.getMagazineIdentifier(), StorageOperation.READ_FIRE_HISTORY); - return client.get(client.getBatchPolicyDefault(), - keys(context).pointerKeys().toArray(new Key[0]), - AerospikeConstants.getFireHistoryBins()); - }); - } - - private Optional resolveCheckpoint(final MagazineContext context, - final Record pointerRecord, - final Instant instant, - final int shard) { - final SortedMap history = readHistory(pointerRecord); - final long cutoff = instant.toEpochMilli(); - for (Map.Entry entry : history.entrySet()) { - final long windowStart = -entry.getKey(); - if (windowStart <= cutoff) { - return Optional.of(new FireCheckpoint(entry.getValue(), Instant.ofEpochMilli(windowStart))); - } - } - // At capacity with nothing old enough means history that could have answered was evicted. - // Returning empty would be indistinguishable from "nothing to do" and would silently stop - // the caller from ever acting. - if (history.size() >= fireHistoryEntries) { - throw MagazineExceptions.invalidRequest(String.format(ErrorMessage.FIRE_HISTORY_TOO_SHORT, - context.getMagazineIdentifier(), shard, instant, - Instant.ofEpochMilli(-history.lastKey()))); - } - return Optional.empty(); - } - - @SuppressWarnings("unchecked") - private SortedMap readHistory(final Record pointerRecord) { - final Object raw = Objects.isNull(pointerRecord) ? null : pointerRecord.getValue(AerospikeConstants.FIRE_HISTORY); - return raw instanceof Map ? new TreeMap<>((Map) raw) : new TreeMap<>(); - } - - /** - * Records one checkpoint per shard, at most once per window per process. - *

- * Deliberately best-effort: a checkpoint improves the reader's precision but is never a - * correctness requirement, since a missed window leaves the previous checkpoint in place and - * that still states something true. Failing a cache refresh or a read because a checkpoint - * could not be written would turn a cosmetic loss into an outage. - *

- * Claimed per shard, and the claim is released again if the write fails, so a shard that had no - * record yet or whose write errored is retried on the next call rather than sitting the window - * out. - */ - private void recordFireHistory(final MagazineContext context, final Record[] pointerRecords) { - if (!fireHistoryEnabled) { - return; - } - final long window = System.currentTimeMillis() / fireHistoryWindowMillis * fireHistoryWindowMillis; - final AtomicLong[] claims = checkpointWindows.computeIfAbsent(context, this::newClaims); - final List pointerKeys = keys(context).pointerKeys(); - // Negated so the key-ordered map sorts newest-first and eviction is a single index range. - final Value windowKey = Value.get(-window); - - for (int shard = 0; shard < context.getShards(); shard++) { - final Record pointers = pointerRecords[shard]; - if (Objects.isNull(pointers) || !claim(claims[shard], window)) { - continue; - } - try { - writeCheckpoint(context, pointerKeys.get(shard), windowKey, - pointers.getLong(AerospikeConstants.FIRE_POINTER)); - } catch (Exception e) { - claims[shard].compareAndSet(window, Long.MIN_VALUE); - log.warn("Could not record fire history for magazine {} shard {} window {}.", - context.getMagazineIdentifier(), shard, window, e); - } - } - } - - private void writeCheckpoint(final MagazineContext context, final Key key, final Value windowKey, - final long firePointer) { - metrics.aerospikeCall(context.getMagazineIdentifier(), StorageOperation.WRITE_FIRE_HISTORY); - client.operate(checkpointPolicy, key, - // CREATE_ONLY|NO_FAIL: the first writer in a window wins and every other process is - // a server-side no-op. Every writer's value is a lower bound, so whichever wins - // recorded the most conservative pointer observed for that window. - MapOperation.put(CHECKPOINT_MAP_POLICY, AerospikeConstants.FIRE_HISTORY, - windowKey, Value.get(firePointer)), - // Keep the newest N by count, not by age: an age cutoff computed from the clock - // would discard the whole map on the first write after an idle period longer than - // the span, blinding readers even though those checkpoints were still usable. - MapOperation.removeByIndexRange(AerospikeConstants.FIRE_HISTORY, - fireHistoryEntries, MapReturnType.NONE)); - } - - private static boolean claim(final AtomicLong claimed, final long window) { - final long previous = claimed.get(); - return previous < window && claimed.compareAndSet(previous, window); - } - - private AtomicLong[] newClaims(final MagazineContext context) { - final AtomicLong[] claims = new AtomicLong[context.getShards()]; - Arrays.setAll(claims, index -> new AtomicLong(Long.MIN_VALUE)); - return claims; + return fireHistory.history(context); } /** @@ -450,6 +260,14 @@ public static com.phonepe.magazine.exception.MagazineException missingMetadata(f context.getMagazineIdentifier(), Objects.isNull(shard) ? 0 : shard)); } + /** Unconsumed slots exist and some hold data. */ + private static boolean isActive(final Record pointers, final Record counters) { + return Objects.nonNull(pointers) + && Objects.nonNull(counters) + && counters.getLong(AerospikeConstants.LOAD_COUNTER) > counters.getLong(AerospikeConstants.FIRE_COUNTER) + && pointers.getLong(AerospikeConstants.LOAD_POINTER) > pointers.getLong(AerospikeConstants.FIRE_POINTER); + } + private void adjustCounter(final MagazineContext context, final Integer shard, final StorageOperation operation, @@ -475,6 +293,21 @@ private Record operateOnMetadata(final Key key, }); } + /** Metadata bins plus fire history's refresh operation, if any. */ + private static Operation[] refreshOperations(final FireHistoryStore fireHistory) { + return Stream.concat(Arrays.stream(AerospikeConstants.getMetadataBins()).map(Operation::get), + fireHistory.refreshReadOperation().stream()) + .toArray(Operation[]::new); + } + + private Record[] refreshRead(final MagazineContext context) { + return retryerFactory.call(() -> { + metrics.aerospikeCall(context.getMagazineIdentifier(), StorageOperation.REFRESH_ACTIVE_SHARDS); + return client.get(client.getBatchPolicyDefault(), + keys(context).pointerKeys().toArray(new Key[0]), refreshOperations); + }); + } + private Record[] batchRead(final MagazineContext context, final List keys, final StorageOperation operation) { diff --git a/magazine-core/src/main/java/com/phonepe/magazine/metrics/MagazineMetrics.java b/magazine-core/src/main/java/com/phonepe/magazine/metrics/MagazineMetrics.java index edcf8b0..8770209 100644 --- a/magazine-core/src/main/java/com/phonepe/magazine/metrics/MagazineMetrics.java +++ b/magazine-core/src/main/java/com/phonepe/magazine/metrics/MagazineMetrics.java @@ -34,7 +34,7 @@ * {@link #aerospikeCall} counts round trips rather than logical operations, which is the * only way to see a dequeue's cost change under contention. *

- * Written to allocate nothing on the hot path: meters are resolved once per magazine into a + * Written to allocate nothing on the hot path: meters are resolved on first use into a * {@link Meters} holder indexed by enum ordinal, so recording is a map lookup plus an array index. * Percentile histograms are deliberately not enabled - they allocate hundreds of buckets per timer * and do bucket arithmetic per record. Add a {@link io.micrometer.core.instrument.config.MeterFilter} @@ -61,7 +61,7 @@ public final class MagazineMetrics { private static final String TAG_OUTCOME = "outcome"; private static final String TAG_OPERATION = "operation"; - /** Lets {@link Meters#outcomes} register any outcome enum without knowing which one it has. */ + /** Lets {@link Meters#outcome} register any outcome enum without knowing which one it has. */ interface Tagged { String getTag(); @@ -119,27 +119,31 @@ public MagazineMetrics(final MeterRegistry registry) { } public void aerospikeCall(final String magazineIdentifier, final StorageOperation operation) { - meters(magazineIdentifier).calls[operation.ordinal()].increment(); + meters(magazineIdentifier).call(operation).increment(); } public void fireClaim(final String magazineIdentifier, final ClaimOutcome outcome) { - meters(magazineIdentifier).fireClaims[outcome.ordinal()].increment(); + final Meters meters = meters(magazineIdentifier); + meters.outcome(meters.fireClaims, FIRE_CLAIMS, outcome).increment(); } public void fireHoleSkip(final String magazineIdentifier) { - meters(magazineIdentifier).fireHoleSkips.increment(); + meters(magazineIdentifier).fireHoleSkips().increment(); } public void fireOutcome(final String magazineIdentifier, final FireOutcome outcome) { - meters(magazineIdentifier).fireOutcomes[outcome.ordinal()].increment(); + final Meters meters = meters(magazineIdentifier); + meters.outcome(meters.fireOutcomes, FIRE_OUTCOMES, outcome).increment(); } public void loadOutcome(final String magazineIdentifier, final LoadOutcome outcome) { - meters(magazineIdentifier).loadOutcomes[outcome.ordinal()].increment(); + final Meters meters = meters(magazineIdentifier); + meters.outcome(meters.loadOutcomes, LOAD_OUTCOMES, outcome).increment(); } public void dedupeOutcome(final String magazineIdentifier, final DedupeOutcome outcome) { - meters(magazineIdentifier).dedupeOutcomes[outcome.ordinal()].increment(); + final Meters meters = meters(magazineIdentifier); + meters.outcome(meters.dedupeOutcomes, DEDUPE_OUTCOMES, outcome).increment(); } /** A primitive rather than a {@code Timer.Sample}, so timing allocates nothing. */ @@ -148,12 +152,12 @@ public long startNanos() { } public void recordFire(final long startNanos, final String magazineIdentifier) { - meters(magazineIdentifier).fireLatency + meters(magazineIdentifier).fireLatency() .record(System.nanoTime() - startNanos, TimeUnit.NANOSECONDS); } public void recordLoad(final long startNanos, final String magazineIdentifier) { - meters(magazineIdentifier).loadLatency + meters(magazineIdentifier).loadLatency() .record(System.nanoTime() - startNanos, TimeUnit.NANOSECONDS); } @@ -161,46 +165,69 @@ private Meters meters(final String magazineIdentifier) { return byMagazine.computeIfAbsent(magazineIdentifier, id -> new Meters(registry, id)); } + /** + * One magazine's meters, registered on first use. A racing first use resolves to the same + * meter in the registry. No allocation once registered. + */ private static final class Meters { - private final Counter[] calls; - private final Counter[] fireClaims; - private final Counter[] fireOutcomes; - private final Counter[] loadOutcomes; - private final Counter[] dedupeOutcomes; - private final Counter fireHoleSkips; - private final Timer fireLatency; - private final Timer loadLatency; + private final MeterRegistry registry; + private final Tags magazine; + private final Counter[] calls = new Counter[StorageOperation.values().length]; + private final Counter[] fireClaims = new Counter[ClaimOutcome.values().length]; + private final Counter[] fireOutcomes = new Counter[FireOutcome.values().length]; + private final Counter[] loadOutcomes = new Counter[LoadOutcome.values().length]; + private final Counter[] dedupeOutcomes = new Counter[DedupeOutcome.values().length]; + private Counter fireHoleSkips; + private Timer fireLatency; + private Timer loadLatency; private Meters(final MeterRegistry registry, final String magazineIdentifier) { - final Tags magazine = Tags.of(TAG_MAGAZINE, magazineIdentifier); - final StorageOperation[] operations = StorageOperation.values(); - this.calls = new Counter[operations.length]; - for (StorageOperation operation : operations) { - this.calls[operation.ordinal()] = Counter.builder(AEROSPIKE_CALLS) + this.registry = registry; + this.magazine = Tags.of(TAG_MAGAZINE, magazineIdentifier); + } + + Counter call(final StorageOperation operation) { + Counter counter = calls[operation.ordinal()]; + if (Objects.isNull(counter)) { + counter = Counter.builder(AEROSPIKE_CALLS) .tags(magazine.and(TAG_OPERATION, operation.getTag())) .register(registry); + calls[operation.ordinal()] = counter; } - this.fireClaims = outcomes(registry, FIRE_CLAIMS, magazine, ClaimOutcome.values()); - this.fireOutcomes = outcomes(registry, FIRE_OUTCOMES, magazine, FireOutcome.values()); - this.loadOutcomes = outcomes(registry, LOAD_OUTCOMES, magazine, LoadOutcome.values()); - this.dedupeOutcomes = outcomes(registry, DEDUPE_OUTCOMES, magazine, DedupeOutcome.values()); - this.fireHoleSkips = Counter.builder(FIRE_HOLE_SKIPS).tags(magazine).register(registry); - this.fireLatency = Timer.builder(FIRE_LATENCY).tags(magazine).register(registry); - this.loadLatency = Timer.builder(LOAD_LATENCY).tags(magazine).register(registry); + return counter; } - private static & Tagged> Counter[] outcomes(final MeterRegistry registry, - final String name, - final Tags magazine, - final E[] values) { - final Counter[] counters = new Counter[values.length]; - for (E outcome : values) { - counters[outcome.ordinal()] = Counter.builder(name) + & Tagged> Counter outcome(final Counter[] slots, final String name, final E outcome) { + Counter counter = slots[outcome.ordinal()]; + if (Objects.isNull(counter)) { + counter = Counter.builder(name) .tags(magazine.and(TAG_OUTCOME, outcome.getTag())) .register(registry); + slots[outcome.ordinal()] = counter; + } + return counter; + } + + Counter fireHoleSkips() { + if (Objects.isNull(fireHoleSkips)) { + fireHoleSkips = Counter.builder(FIRE_HOLE_SKIPS).tags(magazine).register(registry); + } + return fireHoleSkips; + } + + Timer fireLatency() { + if (Objects.isNull(fireLatency)) { + fireLatency = Timer.builder(FIRE_LATENCY).tags(magazine).register(registry); + } + return fireLatency; + } + + Timer loadLatency() { + if (Objects.isNull(loadLatency)) { + loadLatency = Timer.builder(LOAD_LATENCY).tags(magazine).register(registry); } - return counters; + return loadLatency; } } } diff --git a/magazine-core/src/main/java/com/phonepe/magazine/metrics/StorageOperation.java b/magazine-core/src/main/java/com/phonepe/magazine/metrics/StorageOperation.java index 641fabf..edd3a51 100644 --- a/magazine-core/src/main/java/com/phonepe/magazine/metrics/StorageOperation.java +++ b/magazine-core/src/main/java/com/phonepe/magazine/metrics/StorageOperation.java @@ -52,7 +52,7 @@ public enum StorageOperation { DELETE_DATA("delete_data"), BATCH_READ_DATA("batch_read_data"), BATCH_DELETE_DATA("batch_delete_data"), - /** The checkpoint write; one per shard per window, not one per claim. */ + /** Checkpoint write: one batch call per magazine, for shards whose pointer moved. */ WRITE_FIRE_HISTORY("write_fire_history"), /** The batch read behind {@code firePointerBefore}. */ READ_FIRE_HISTORY("read_fire_history"), diff --git a/magazine-core/src/test/java/com/phonepe/magazine/impl/aerospike/MagazineFireHistoryTest.java b/magazine-core/src/test/java/com/phonepe/magazine/impl/aerospike/MagazineFireHistoryTest.java index 74f04f3..94a98bb 100644 --- a/magazine-core/src/test/java/com/phonepe/magazine/impl/aerospike/MagazineFireHistoryTest.java +++ b/magazine-core/src/test/java/com/phonepe/magazine/impl/aerospike/MagazineFireHistoryTest.java @@ -24,6 +24,7 @@ import com.phonepe.magazine.exception.ErrorCode; import com.phonepe.magazine.exception.MagazineException; import com.phonepe.magazine.impl.aerospike.common.AerospikeConstants; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import org.awaitility.Awaitility; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; @@ -68,6 +69,7 @@ void disabledByDefault() { void recordsOneCheckpointPerWindow() { final int shards = 4; // A wide window so the whole test runs inside one, making "same window" unambiguous. + awaitFreshWindow(60); final Magazine magazine = magazine("HIST_ONE", "HIST_ONE_SET", true, shards, 8, 60); loadAndFire(magazine, 3); @@ -146,14 +148,32 @@ void returnsNewestCheckpointNotAfterTheInstant() { loadAndFire(magazine, 4); magazine.firePointerBefore(Instant.now()); - final FireCheckpoint checkpoint = magazine.firePointerBefore(afterFirstBurst).get("SHARD_0"); + // Checkpoints are labelled with the end of the window they were observed in, so the first + // burst's checkpoint becomes answerable one window after it was taken - and the second + // burst, which started later still, cannot be. + final Instant question = afterFirstBurst.plusSeconds(WINDOW_SECONDS); + final FireCheckpoint checkpoint = magazine.firePointerBefore(question).get("SHARD_0"); assertNotNull(checkpoint, "the first burst's window is old enough to answer with"); - assertFalse(checkpoint.recordedAt().isAfter(afterFirstBurst), + assertFalse(checkpoint.recordedAt().isAfter(question), "the answer must come from a window that had already closed by then"); assertTrue(checkpoint.firePointer() <= pointerAfterFirstBurst, "the second burst's pointer must not be used to answer an earlier question"); } + @Test + @DisplayName("a checkpoint's label never precedes the moment its pointer was observed") + void labelNeverPrecedesObservation() { + final Magazine magazine = magazine("HIST_LABEL", "HIST_LABEL_SET", true, 1, 8); + loadAndFire(magazine, 2); + final long beforeRecording = System.currentTimeMillis(); + magazine.firePointerBefore(Instant.now()); + + // Labelling with the window start would claim the pointer was reached before it was read, + // which over-reports by however far into the window the write happened. + assertTrue(-newestWindowKey(magazine) >= beforeRecording, + "the newest label must be at or after the observation it records"); + } + @Test @DisplayName("a young magazine reports nothing rather than failing") void youngHistoryReturnsAbsent() { @@ -187,24 +207,83 @@ void evictedHistoryFailsLoudly() { @Test @DisplayName("reading advances history, so a burst that is never fired from again stays reachable") void readingRecordsACheckpoint() { - final Magazine magazine = magazine("HIST_BURST", "HIST_BURST_SET", true, 1, 8); + // Refresh effectively off, so only the read can be the one recording. + final Magazine magazine = magazine("HIST_BURST", "HIST_BURST_SET", true, 1, 8, + WINDOW_SECONDS, 3600, null); loadAndFire(magazine, 3); + final long firePointer = metadataRecord(magazine, 0).getLong(AerospikeConstants.FIRE_POINTER); + + awaitNextWindow(); + // No load, no fire: the magazine is idle. Reading alone must bring history up to the + // current pointer, otherwise a magazine that took one burst and was then abandoned would + // hold a checkpoint predating the burst and its orphans would never become visible. magazine.firePointerBefore(Instant.now()); - final long newestBefore = newestWindowKey(magazine); + assertEquals(firePointer, history(magazine, 0).get(newestWindowKey(magazine)), + "the newest checkpoint should hold the pointer the burst left behind"); + } + + @Test + @DisplayName("an idle shard records nothing new, from the refresh or from reads") + void idleShardWritesNothing() { + final SimpleMeterRegistry registry = new SimpleMeterRegistry(); + final Magazine magazine = magazine("HIST_IDLE", "HIST_IDLE_SET", true, 1, 8, + WINDOW_SECONDS, 1, registry); + loadAndFire(magazine, 2); + // A refresh during the burst may have claimed this window with a mid-burst pointer, in + // which case the final pointer is recorded in the next window. Settle that first. + magazine.firePointerBefore(Instant.now()); awaitNextWindow(); - // No load, no fire: the magazine is idle. Reading alone must still move history forward, - // otherwise a magazine that took one burst and was then abandoned would hold a single - // checkpoint predating the burst and its orphans would never become visible. magazine.firePointerBefore(Instant.now()); + final int entries = history(magazine, 0).size(); + final double writes = writeCalls(registry); + + for (int window = 0; window < 3; window++) { + awaitNextWindow(); + magazine.firePointerBefore(Instant.now()); + } - // The property, not a count. The active-shard refresh records on its own schedule, so how - // many windows exist by this point is not something the test controls - under load the - // preceding loadAndFire can contribute one of its own. What the read must guarantee is that - // a window strictly newer than anything present beforehand now exists. - assertTrue(newestWindowKey(magazine) < newestBefore, - "the read should have recorded a newer window by itself; keys are negated window " - + "timestamps, so newest sorts lowest"); + // The fire pointer only moves forward, so a checkpoint that already holds the current + // value stays true for every later instant. Rewriting it would cost a call per shard per + // window per process for magazines that are doing nothing at all. + assertEquals(entries, history(magazine, 0).size(), "no new checkpoint for an unchanged pointer"); + assertEquals(writes, writeCalls(registry), "no write round trip for an unchanged pointer"); + } + + @Test + @DisplayName("a second process skips a window another process already recorded") + void secondProcessSkipsARecordedWindow() { + awaitFreshWindow(60); + final Magazine first = magazine("HIST_TWO", "HIST_TWO_SET", true, 1, 8, 60, 3600, null); + loadAndFire(first, 3); + first.firePointerBefore(Instant.now()); + + final SimpleMeterRegistry registry = new SimpleMeterRegistry(); + final Magazine second = magazine("HIST_TWO", "HIST_TWO_SET", true, 1, 8, 60, 3600, registry); + second.firePointerBefore(Instant.now()); + + assertEquals(0.0, writeCalls(registry), + "the window's checkpoint is already on the record, so the second process must not call"); + } + + @Test + @DisplayName("every shard's checkpoint goes out in one batch call") + void checkpointsAreBatched() { + final int shards = 8; + final SimpleMeterRegistry registry = new SimpleMeterRegistry(); + awaitFreshWindow(60); + final Magazine magazine = magazine("HIST_BATCH", "HIST_BATCH_SET", true, shards, 8, + 60, 3600, registry); + for (int i = 0; i < 40; i++) { + magazine.load("message-" + random.nextLong()); + } + final int withRecords = shardsWithMetadata(magazine, shards); + assertTrue(withRecords > 1, "expected several shards to have been written to"); + + magazine.firePointerBefore(Instant.now()); + + assertEquals(withRecords, totalCheckpoints(magazine, shards)); + assertEquals(1.0, writeCalls(registry), "one round trip for the whole magazine, not one per shard"); } @Test @@ -218,7 +297,8 @@ void firingRecordsCheckpoints() { Awaitility.await().atMost(Duration.ofSeconds(20)).pollInterval(Duration.ofMillis(200)) .untilAsserted(() -> { loadAndFire(magazine, 1); - assertFalse(history(magazine, 0).isEmpty(), + final Map history = history(magazine, 0); + assertTrue(history != null && !history.isEmpty(), "firing alone should eventually record a checkpoint"); }); } @@ -228,6 +308,7 @@ void firingRecordsCheckpoints() { void lateShardIsCheckpointedInTheSameWindow() { final int shards = 8; // One window wide enough to hold the whole test, so nothing here can roll into the next. + awaitFreshWindow(60); final Magazine magazine = magazine("HIST_LATE", "HIST_LATE_SET", true, shards, 8, 60); loadAndFire(magazine, 1); magazine.firePointerBefore(Instant.now()); @@ -313,12 +394,26 @@ private Magazine magazine(final String identifier, final String setName, private Magazine magazine(final String identifier, final String setName, final boolean enabled, final int shards, final int entries, final int windowSeconds) { + return magazine(identifier, setName, enabled, shards, entries, windowSeconds, 1, null); + } + + private Magazine magazine(final String identifier, final String setName, + final boolean enabled, final int shards, final int entries, + final int windowSeconds, final int refreshSeconds, + final SimpleMeterRegistry registry) { return Magazine.builder() .magazineIdentifier(identifier) - .baseMagazineStorage(storageWithHistory(setName, enabled, windowSeconds, entries, shards)) + .baseMagazineStorage(buildStorage(config(setName, enabled, windowSeconds, entries, shards, + refreshSeconds), + String.class, false, "FARM_ID", "CLIENT_ID", MagazineScope.LOCAL, aerospikeClient, + registry)) .build(); } + private static double writeCalls(final SimpleMeterRegistry registry) { + return counterValue(registry, "magazine.aerospike.calls", "operation", "write_fire_history"); + } + private AerospikeStorage storageWithHistory(final String setName, final boolean enabled, final int windowSeconds, final int entries) { return storageWithHistory(setName, enabled, windowSeconds, entries, 1); @@ -327,7 +422,14 @@ private AerospikeStorage storageWithHistory(final String setName, final private AerospikeStorage storageWithHistory(final String setName, final boolean enabled, final int windowSeconds, final int entries, final int shards) { - return buildStorage(AerospikeStorageConfig.builder() + return buildStorage(config(setName, enabled, windowSeconds, entries, shards, 1), + String.class, false, "FARM_ID", "CLIENT_ID", MagazineScope.LOCAL, aerospikeClient); + } + + private static AerospikeStorageConfig config(final String setName, final boolean enabled, + final int windowSeconds, final int entries, + final int shards, final int refreshSeconds) { + return AerospikeStorageConfig.builder() .namespace(NAMESPACE) .dataSetName(setName) .metaSetName("META_SET") @@ -335,12 +437,25 @@ private AerospikeStorage storageWithHistory(final String setName, final .metaDataTtl(2 * 30 * 24 * 60 * 60) .shards(shards) // Refresh often, so the fire path exercises the recorder within a test's life. - .activeShardRefreshSeconds(1) + .activeShardRefreshSeconds(refreshSeconds) .fireHistoryEnabled(enabled) .fireHistoryWindowSeconds(windowSeconds) .fireHistoryEntries(entries) - .build(), - String.class, false, "FARM_ID", "CLIENT_ID", MagazineScope.LOCAL, aerospikeClient); + .build(); + } + + /** + * Windows are aligned to the clock, so a test that must stay inside one window starts near the + * beginning of one rather than wherever the clock happens to be. + */ + private static void awaitFreshWindow(final int windowSeconds) { + final long windowMillis = windowSeconds * 1_000L; + final long remaining = windowMillis - System.currentTimeMillis() % windowMillis; + // These tests take a couple of seconds; fifteen leaves ample margin. + if (remaining < 15_000) { + Awaitility.await("the start of a fresh window").pollDelay(Duration.ofMillis(remaining + 50)) + .atMost(Duration.ofMillis(remaining + 5_000)).until(() -> true); + } } private void awaitNextWindow() { diff --git a/magazine-core/src/test/java/com/phonepe/magazine/impl/aerospike/MagazineMetricsTest.java b/magazine-core/src/test/java/com/phonepe/magazine/impl/aerospike/MagazineMetricsTest.java index d36563f..8961aab 100644 --- a/magazine-core/src/test/java/com/phonepe/magazine/impl/aerospike/MagazineMetricsTest.java +++ b/magazine-core/src/test/java/com/phonepe/magazine/impl/aerospike/MagazineMetricsTest.java @@ -134,4 +134,28 @@ void metricsCanBeDisabledExplicitlyEvenWithARegistry() { assertEquals(0.0, counterValue(registry, "magazine.aerospike.calls", "operation", "write_data")); assertTrue(registry.getMeters().isEmpty(), "disabled metrics must register no meters"); } + + /** + * A meter appears the first time it is recorded. Registering every operation and outcome up + * front meant thirty meters per magazine, most of them zero for the life of the process. + */ + @Test + void onlyMetersThatHaveRecordedSomethingAreRegistered() { + SimpleMeterRegistry registry = new SimpleMeterRegistry(); + AerospikeStorage storage = buildStorage( + buildStorageConfig("NAMESPACE", "DATA_SET", "META_SET", + 30 * 24 * 60 * 60, 2 * 30 * 24 * 60 * 60, 1), + String.class, false, "FARM_ID", "CLIENT_ID", MagazineScope.LOCAL, aerospikeClient, + registry); + MagazineContext context = storage.initialize("LAZY_METERS_MAGAZINE"); + + assertTrue(storage.load(context, "PAYLOAD")); + + assertEquals(1.0, counterValue(registry, "magazine.aerospike.calls", "operation", "write_data")); + assertTrue(registry.find("magazine.aerospike.calls").tag("operation", "claim_fire_pointer") + .counters().isEmpty(), "nothing has been fired, so the claim counter must not exist"); + assertTrue(registry.find("magazine.fire.latency").timers().isEmpty()); + assertTrue(registry.find("magazine.dedupe.outcomes").counters().isEmpty()); + assertTrue(registry.find("magazine.load.outcomes").tag("outcome", "failed").counters().isEmpty()); + } } diff --git a/magazine-coverage/pom.xml b/magazine-coverage/pom.xml index 749c42a..732275d 100644 --- a/magazine-coverage/pom.xml +++ b/magazine-coverage/pom.xml @@ -7,7 +7,7 @@ com.phonepe magazine - 2.0.1-SNAPSHOT + 2.0.2-SNAPSHOT magazine-coverage diff --git a/magazine-dw-bundle/pom.xml b/magazine-dw-bundle/pom.xml index a304fbd..bd65519 100644 --- a/magazine-dw-bundle/pom.xml +++ b/magazine-dw-bundle/pom.xml @@ -7,7 +7,7 @@ com.phonepe magazine - 2.0.1-SNAPSHOT + 2.0.2-SNAPSHOT magazine-dw-bundle diff --git a/pom.xml b/pom.xml index 710b983..4bf6755 100644 --- a/pom.xml +++ b/pom.xml @@ -7,7 +7,7 @@ com.phonepe magazine - 2.0.1-SNAPSHOT + 2.0.2-SNAPSHOT pom Magazine