Zero-loss message persistence is in-memory only: queued messages are lost on restart

[ameeribro4-sudo]

Labels / Complexity: bug · High — 700 points

Problem

ZeroLossMessageService (src/queue/zero-loss-message.service.ts) claims to guarantee that no messages are lost through "persistence, replication, and acknowledgment", but all of its state lives in process memory:

// src/queue/zero-loss-message.service.ts
private persistedMessages: Map<string, PersistedMessage> = new Map();
private pendingAcknowledgments: Map<string, NodeJS.Timeout> = new Map();
private replicationNodes: Set<string> = new Set();

persistMessage() stores the message in persistedMessages, and registerReplicationNode() adds a node ID to an in-memory Set. The configuration even declares a replicated persistence level that the implementation does not provide:

// src/queue/horizontal-scaling.config.ts
zeroLoss: {
  enabled: true,
  persistenceLevel: 'replicated',   // ← claims replication
  replicationFactor: 3,             // ← claims 3 replicas
  acknowledgmentTimeoutMs: 5000,
  maxRetryAttempts: 5,
},

The consequence is silent data loss. On any process restart, crash, or horizontal scale-out, every "persisted" message, every replication node, and every pending acknowledgment timeout vanishes. getReplicationTargets() returns nodes from the in-memory Set, so replicationNodes.length is always 0 on a fresh instance, and verifyMessageIntegrity() will report Insufficient replication: 0/3 for every message. A message marked processing by a worker that then dies is never re-queued, because the timeout that would requeue it only exists in the dead process's memory.

This matters because the whole HorizontalScalingModule (src/queue/horizontal-scaling.module.ts) advertises zero-loss and multi-instance behavior, yet the state is per-process and non-durable.

Root cause

// src/queue/zero-loss-message.service.ts
this.persistedMessages.set(messageId, message);   // ← in-memory only, lost on restart

Why this is architecturally hard

1. The natural shortcut fails. A bigger in-memory Map (or moving it to a module-level singleton) does not survive a restart and does not share state across instances, which is the entire point of a zero-loss layer on a horizontally-scaled queue.
2. A real backend must be chosen. The durable store has to be Redis (already used by Bull via src/queue/queue.module.ts and by RedisPoolService under src/common/cache/) or Postgres (via TypeORM). Each has different semantics for TTLs, atomicity, and the acknowledgment timeout.
3. Recovery of orphaned processing messages is required. A boot-time or periodic sweep must find messages stuck in processing past the acknowledgment timeout and requeue them, without double-processing messages whose worker is actually still alive. This needs a lock/lease, not a timestamp check.
4. The design must not duplicate Bull. Bull already persists jobs in Redis with its own retry and stalled-job handling (stalledInterval is configured in queue.module.ts). The contributor must decide whether ZeroLossMessageService should back onto Bull's own durability, or maintain a parallel store, and justify that choice.

Proposed design

Replace the in-memory Map/Set with a durable store keyed by messageId. A reasonable target signature keeps the public API stable:

persistMessage(messageId, queueName, data, maxAttempts?): Promise<PersistedMessage>
markProcessing(messageId): Promise<boolean>
acknowledgeMessage(messageId): Promise<boolean>

Use Redis hashes or strings with a TTL for the acknowledgment window, or a PersistedMessage table with a status + updatedAt index for the recovery sweep. Replication nodes should be persisted (a set key) so getReplicationTargets() returns the same targets across instances. Add a recovery path that atomically transitions processing → pending only when the lease has expired.

Acceptance criteria

Service

• After a simulated crash and restart, a message persisted before the crash is still present and can be re-queued.
• A message stuck in processing past the acknowledgment timeout is transitioned to pending exactly once (no double-processing).
• replicationNodes survive a restart and are shared across instances of the same deployment.

Tests

• A test persists a message, simulates process restart by reinstantiating the service against the same store, and asserts the message is recoverable.
• A test verifies the processing → pending requeue fires once for an orphaned message.

Documentation

• The chosen durability backend, its failure modes, and the recovery sweep interval are documented in the queue module or README.

Out of scope

Do not change deduplication (src/queue/message-deduplication.service.ts) or ordering (src/queue/message-ordering.service.ts) in this issue; those are adjacent but separate in-memory-state problems.

Getting started

Files in scope: src/queue/zero-loss-message.service.ts, src/queue/horizontal-scaling.config.ts, src/queue/horizontal-scaling.module.ts, and the Redis utilities under src/common/cache/.

Build and test with:

npm run build
npm run test

Good first files to read: src/queue/dead-letter-queue.service.ts (a sibling service with the same in-memory problem), src/common/cache/redis-pool.service.ts (the existing Redis accessor).

[ayomidearegbeshola29-dev]

I’d like to take this issue on.

Approach

I’ll first trace the existing queue architecture, especially:

• ZeroLossMessageService
• horizontal-scaling.module.ts
• queue.module.ts
• RedisPoolService
• Bull configuration and stalled-job handling
• dead-letter-queue.service.ts
• Existing queue tests and Redis test infrastructure

The key design decision I’ll make before implementation is whether the zero-loss service should use Bull's existing Redis persistence or maintain a small dedicated durable Redis state layer.

My initial preference is to use the existing Redis infrastructure while keeping the zero-loss state separate from Bull's job lifecycle unless the repository already exposes enough Bull primitives to safely model the required persistence/lease semantics. I don't want to duplicate retry or stalled-job functionality that Bull already provides.

Durable State

The current process-local structures:

persistedMessages: Map
pendingAcknowledgments: Map
replicationNodes: Set

will be replaced with shared durable Redis state.

The public service API will remain asynchronous and compatible with the existing callers:

persistMessage(...)
markProcessing(...)
acknowledgeMessage(...)

Messages will contain enough state to recover safely, including their status, attempt information, timestamps/lease information, queue name, and payload.

Replication nodes will also live in Redis so separate service instances see the same registered nodes.

Processing / Recovery Model

The important part of this issue is avoiding a simple timestamp-based requeue that could result in duplicate processing.

I'll model processing as a lease:

1. A worker atomically claims a message and receives/renews its processing lease.
2. Successful processing acknowledges the message and removes/releases its durable state.
3. If the worker disappears, the lease expires.
4. A recovery sweep finds expired processing leases.
5. The transition from processing → pending is performed atomically so multiple application instances cannot recover the same message simultaneously.
6. The message can then be submitted back to the queue.

This gives the recovery mechanism a clear ownership boundary rather than relying on an in-memory setTimeout.

I'll also compare this behavior against Bull's existing stalled-job/retry mechanisms so the two systems don't independently retry the same work.

Crash and Horizontal-Scaling Tests

I'll add integration-style tests using the same durable store across separate service instances.

The key scenarios will include:

• Instance A persists a message.
• Instance A is discarded/restarted.
• Instance B connects to the same Redis store.
• The message remains available and can be recovered.
• A message is marked processing.
• Its lease expires.
• The recovery process transitions it to pending.
• Multiple recovery attempts/instances cannot requeue it more than once.
• Registered replication nodes remain visible after creating a new service instance.
• Multiple service instances observe the same replication state.

I'll also test successful acknowledgment so recovered messages don't remain eligible for reprocessing after being completed.

Failure Handling

I'll explicitly account for Redis/backend failures rather than claiming zero-loss when the durable backend itself is unavailable.

The implementation will distinguish between:

• message not found,
• already acknowledged,
• active processing lease,
• expired processing lease,
• persistence failure,
• replication/recovery failure.

Where appropriate, operations will fail visibly rather than silently falling back to the old in-memory behavior. I won't retain an in-memory fallback that could make the configured replicated durability level appear to work while still losing messages on restart.

Documentation

I'll document:

• The selected durability backend.
• What state is persisted.
• Replication-node persistence.
• Processing lease semantics.
• Recovery sweep interval.
• What happens when Redis is unavailable.
• How this interacts with Bull's own retry/stalled-job handling.
• The exact guarantees the service provides versus guarantees dependent on Redis availability.

I'll also verify that horizontal-scaling.config.ts accurately reflects the implementation rather than advertising replication that only exists in process memory.

Validation

I'll run the existing build and test suite:

npm run build
npm run test

and add targeted tests for crash recovery, lease expiry, atomic requeue, acknowledgment, and cross-instance visibility.

I'll keep message-deduplication.service.ts and message-ordering.service.ts untouched as required.

The implementation will avoid introducing a second queue/retry mechanism where Bull already provides the functionality, and I'll keep the Redis operations atomic wherever state transitions could otherwise race between horizontally-scaled instances.

The PR will explain the durability/lease design, Bull integration decision, failure modes, and test evidence, while keeping the scope limited to the zero-loss message layer.

[1Judah]

Hello please assign me this issue. My initial preference is to use the existing Redis infrastructure while keeping the zero-loss state separate from Bull's job lifecycle unless the repository already exposes enough Bull primitives to safely model the required persistence/lease semantics. I don't want to duplicate retry or stalled-job functionality that Bull already provides.

[grantfox-oss]

🦊 GrantFox — @1Judah has been assigned to this issue as part of the Third Campaign campaign!

Next steps:

1. Open a Pull Request referencing this issue (e.g., Closes #14)
2. Your PR will be reviewed by the OpenPeerX maintainers

Good luck! Track your progress on GrantFox.

[grantfox-oss]

🎉 This issue has been marked as completed on GrantFox as part of the Third Campaign campaign!

@1Judah's PR #21 was approved and merged by @ameeribro4-sudo.

🏆 @1Judah: You earned 35 FoxPoints for this contribution! Your current tier: Explorer (239 total points). Track your full progress on GrantFox.

👏 Great work, @1Judah! Keep contributing to OpenPeerX.