Aerospike Backend¶

Comprehensive documentation of the Aerospike storage backend for IgnisMQ.

Overview¶

IgnisMQ uses Aerospike as its primary storage backend, leveraging the Magazine library for queue primitives (load/fire operations) and AerospikeQueueService for queue metadata management.

The storage layer is split into two concerns:

• AerospikeStorage — a thin configuration holder that carries the AerospikeConfiguration and namespace, implementing the BaseStorage visitor pattern.
• AerospikeQueueService — the CRUD for queue metadata and the sweep pointers. It is not on the message path: message slots are read and written through Magazine's own storage, not through this class.

Configuration¶

AerospikeStorage storage = new AerospikeStorage(
    AerospikeConfiguration.builder()
        .hosts(List.of(AerospikeHost.builder()
            .host("localhost")
            .port(3000)
            .build()))
        .retries(3)
        .sleepBetweenRetries(100)
        .socketTimeout(3000)
        .totalTimeout(5000)
        .maxConnectionsPerNode(100)
        .threadPoolSize(4)   // default: availableProcessors * 4
        .build(),
    "my-namespace"
);

Queue Metadata Set: {farmId}_{clientId}_ignis_queues¶

Each record represents one queue. The Aerospike key is the queue name.

Magazine Data Sets (managed by Magazine library)¶

Secondary Indexes¶

A secondary index is automatically created on the active bin (IndexType.STRING) during AerospikeQueueService construction:

// Index name: {farmId}_{clientId}_ignis_queues_active
createIndex(setName + "_" + ACTIVE_BIN, ACTIVE_BIN, IndexType.STRING);

This enables efficient queries for active or inactive queues via Filter.equal(ACTIVE_BIN, "true").

Client Configuration¶

The Aerospike client is configured in AerospikeStoreClient with the following policies:

Retry Strategy¶

All AerospikeQueueService operations use a Guava Retryer:

RetryerBuilder.newBuilder()
    .retryIfExceptionOfType(AerospikeException.class)
    .withStopStrategy(StopStrategies.stopAfterAttempt(configuration.getRetries()))
    .withWaitStrategy(WaitStrategies.fixedWait(
        configuration.getSleepBetweenRetries(), TimeUnit.MILLISECONDS))
    .withBlockStrategy(BlockStrategies.threadSleepStrategy())
    .build();

Delivery-Time Watermarks¶

The sweeper has to answer one question: was this message handed to a consumer long enough ago that the consumer is certainly gone? The old answer was per message — a timestamp on every record. The current answer is positional, and costs nothing per message.

Magazine periodically records where each shard's fire pointer stood at a point in time, a checkpoint. Given a cutoff, firePointerBefore(Instant) returns, per shard, the highest fire pointer known to have been reached at or before that moment:

Map<String, FireCheckpoint> watermarks =
        magazine.firePointerBefore(Instant.ofEpochMilli(now - sweepDurationMillis));

Everything at or below that pointer was claimed at least sweepDuration ago. Everything above it might have been claimed a moment ago and must not be touched.

How checkpoints are recorded¶

IgnisMQ uses a window of sweepDuration / 8 (150 seconds at the default 20 minutes) and keeps 32 checkpoints per shard, so history reaches back four sweep durations. Magazine records them from its active-shard refresh:

• At most one checkpoint per shard per window, across all instances. Each instance waits a random offset into the first half of the window; the first to get there writes, the others see it and skip.
• None for a shard whose pointer has not moved. Idle queues cost nothing.
• One batch call per magazine, not one per shard.

The sweeper, which runs only on the leader, also records when it reads. This normally writes nothing because the refresh already has; it matters when consumption has stopped everywhere, so the last claims still become sweepable.

A checkpoint is labelled with the end of the window it was taken in, so it can be used one window later. A message abandoned at time T is therefore eligible from about T + sweepDuration + one window, and is moved on the next sweep pass after that.

What this buys¶

Two behaviours worth knowing¶

A shard missing from the answer means "sweep nothing here". That is normal for a young queue — no checkpoint reaches back that far yet — and it is the safe direction.

The sweeper refuses loudly rather than guessing. If the retained checkpoint history no longer spans the requested cutoff, firePointerBefore throws INVALID_REQUEST, the pass is abandoned for that magazine with an error log, and ignismq.sweep{outcome=failure} is recorded. Refusing is the only safe answer — without history the sweeper cannot tell an abandoned message from a live one — and the noise is deliberate: silently sweeping nothing looks exactly like a healthy idle queue.

Checkpoint retention is sized from sweepDuration, so this should not happen in normal operation. Alert on it — see the monitoring runbook.

Sweep Internals¶

Runs on the one instance the leader assigned the sweep partition to, and only for active queues.

1. Sweeper.run() — skipped entirely unless this node holds the sweep assignment.
2. Active queues are fetched once via queueService.getQueues(true).
3. Each queue is submitted to a pool capped at Constants.PARALLEL_FACTOR (64).
4. Both magazines are swept — the main queue and its sideline.

Per magazine¶

flowchart TD
    A[sweepMagazine] --> B["firePointerBefore(now - sweepDuration)"]
    B --> C{Any shard has a checkpoint?}
    C -- No --> Z[Done — nothing is provably old enough]
    C -- Yes --> D[Per shard: limit = checkpoint fire pointer]
    D --> E[Per shard: start = stored sweep pointer]
    E --> F{start <= limit?}
    F -- No --> Z2[Shard already swept to the watermark]
    F -- Yes --> G["peek() a window of pointers, batched across shards"]
    G --> H{Slot still holds a record?}
    H -- No --> I[Delivered and acknowledged — nothing to do]
    H -- Yes --> J[Load into the sideline magazine]
    J --> K{Load succeeded?}
    K -- No --> L[Leave it. The next pass retries]
    K -- Yes --> M[Delete from the source magazine]
    M --> N[Advance the shard's sweep pointer]
    I --> N
    N --> O{Reached the watermark?}
    O -- No --> G
    O -- Yes --> Z3[Done]

Key details¶

• The watermark bounds the range; it does not say what is in it. A slot below the fire pointer is either empty — delivered and acknowledged — or still full, meaning abandoned. Only looking can tell, so the range is still scanned, but with one peek per batch rather than a read per message.
• Every slot is scanned once, ever. The per-shard sweep pointer only moves forward.
• peek is batched across shards, not per shard, so a queue with 8 shards issues one multi-key read per round rather than 8.
• Re-homing is at-least-once, deliberately. The message is loaded into the sideline and only then deleted from the source. A crash between the two leaves it in both, and the next pass re-homes it again. The opposite ordering loses messages.
• A failed load is not a delete. The record stays where it is for a later pass. This is the same contract the consumer follows when the sideline refuses a message.
• Shard count comes from magazine.getShards(), the persisted value, so ignisMQ's view and Magazine's can no longer disagree.
• sweepDuration is capped at 12 hours, and the handler timeout is clamped to at most half of it so that a handler cannot still be running when the sweeper re-homes the record it is working on.

Performance Tips¶