You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
When logmq processes a batch, it first saves all the log entries in one batched write. That part is fine. The problem is what happens next: for each entry, it runs alert evaluation and then sends an opevent, one entry at a time, in a single goroutine. The slowest step is sending the opevent to the customer's sink, so one slow sink slows the whole loop down, including saving new logs.
Proposal: take that post-save work off the saving path so it can't block saving, and run it in parallel instead of one at a time. Split it into two parts:
Alert evaluation — fast (just Redis), and must stay roughly in order per destination.
Opevent delivery — slow (the sink call), order doesn't matter, so run many at once.
Both run in memory. We don't add any new queue or infra. We keep the logmq message un-acked until delivery finishes, so nothing is lost.
What we need to get right (please push back here)
If you disagree with one of these, the design changes.
1. Keep per-destination order in alert eval — roughly, not perfectly. (Order doesn't matter for opevent delivery.)
Why order matters for alert eval, and why we only aim for "roughly"
Alerting counts how many times a destination failed in a row. A success resets that count. So if an old success gets processed in the middle of a run of failures, the count resets at the wrong time and the alert (or auto-disable) can fire late, or get missed. So the order we process attempts in does matter.
But we can't promise perfect order, and we don't today. The order is already a bit loose before logmq even sees it: publishmq → deliverymq → logmq uses parallel workers and retries, so attempts for one destination can show up slightly out of order. Whatever order logmq receives, that's what it works with.
So the goal is: stay close to the real order per destination, like it is today — not perfect. Today everything runs one at a time, so it's roughly in order. The thing we must avoid: if we just run everything fully in parallel, the order per destination becomes random. That's the part we want to prevent. The fix is to always send attempts for the same destination to the same worker, so they stay in order. Different destinations run fully in parallel — their order relative to each other never mattered.
2. A slow opevent delivery must not slow down saving logs.
Where the slow part actually is
The numbers below are rough estimates, not measured — we should confirm with real data before locking in worker counts. Basis: eval is a couple of Redis round-trips (~2ms); a sink call is an outbound HTTP request to a customer endpoint (tens of ms typical), and we'd set the timeout ourselves (~5s is the order of magnitude).
Alert evaluation is fast — roughly a couple of milliseconds, just a couple of Redis calls (count the failures, check the threshold, disable the destination if needed).
The slow part is sending the opevent to the sink — tens of milliseconds normally. We can set an aggressive timeout (say a few seconds) to cap the worst case, but done serially even that becomes a huge bottleneck: if a sink has an issue, every send waits out the timeout, one after another. Today that send happens inside the same one-at-a-time loop that the pipeline waits on. So when one customer's sink is slow, the loop slows down, the batch processor slows down, logmq stops pulling new messages, and saving logs slows down too. One slow sink drags down the whole pipeline. It can also push messages past their visibility timeout while they wait, so the broker redelivers them and we reprocess the same work.
Note: alert evaluation still has to keep up with how fast we save logs, but that's easy — it's fast, so we just run a few of them at once.
3. No new durable queue (no opeventsmq).
Why we rely on the logmq message instead
Sending an opevent should work most of the time. For the rare failure, we don't ack the logmq message — it gets redelivered and we redo the work. That's safe because the whole pipeline is idempotent: re-inserting the log doesn't create a duplicate, and re-running alert eval doesn't double-count or double-emit. So we can nack freely. Adding a separate durable queue just for opevents isn't worth the extra complexity. The in-memory parts below are only for running things in parallel, not for durability.
Proposal
After the batched save, dispatch each entry into two in-memory stages instead of the one-at-a-time loop. The message stays un-acked until delivery finishes (ack on success, nack to retry on failure). The two stages have different shapes: Stage 1 is sharded (keeps order per destination), Stage 2 is shared (parallelism).
flowchart LR
BP["batcher (1 goroutine)<br/>InsertMany (save batch)"] -->|"per entry, routed by hash(destID)"| Q1
BP -.->|"unlocks after dispatch,<br/>doesn't wait for eval/delivery"| NEXT(["next batch starts"])
subgraph S1["Stage 1 — opevent eval (ordered per destination)"]
Q1["N queues<br/>sharded by destination"] --> W1["eval worker 1"]
Q1 --> WN["eval worker N"]
end
W1 -->|opevents| Q2
WN -->|opevents| Q2
subgraph S2["Stage 2 — delivery (unordered)"]
Q2["1 shared queue"] --> D1["delivery worker 1"]
Q2 --> DM["delivery worker M"]
end
D1 --> ACK["ack / nack logmq msg"]
DM --> ACK
Loading
Design assumption: one opevent per log entry is the expected case, not a rare spike — we intend to support opevents generated per entry. So we size and reason as if each entry generates one; we don't lean on "most entries emit nothing."
Freeing the batcher. The batcher returns as soon as it has dispatched the batch's entries into Stage 1; it does not wait for eval or delivery. That hand-off is what unlocks the next batch, and it's the core of the decoupling. (Today the batcher is held through the whole serial eval+send loop, so the next batch waits on the slowest entry.)
Stage 1 — opevent evaluation. This stage decides what opevents an entry produces. Same destination always goes to the same worker, so attempts stay in order per destination, and each one is fast (~2ms). Today every opevent here comes from alert eval (count failures in Redis, check the threshold, disable the destination if needed → consecutive-failure, disabled, exhausted-retries). But opevents aren't inherently alerts; framing the stage as "produce this entry's opevents" rather than "alert eval" means a future non-alert opevent fits here without reshaping the pipeline. The ordering requirement comes from alert eval specifically (its failure count is order-sensitive); any other producer just rides the same per-destination sharding. It produces this entry's opevents (0 to 3), hands the entry to Stage 2, and moves on — it does not wait for the send. If the entry produced no opevent, ack the message right away.
Stage 2 — opevent delivery. One delivery task per entry (not per opevent): a worker sends that entry's opevents together, then acks/nacks the one message behind them. Any worker can pick up any entry; we run many entries at once. This is where the slow sink call lives, off to the side. One slow destination can't back things up here, because delivery isn't tied to a destination.
Acking. Each message has exactly one delivery task, so there's no fan-in to track: ack when that task's sends all succeed, nack if any fail (it redelivers and redoes the entry). If the process crashes, the message was never acked, so the broker redelivers it — nothing is lost.
Replay safety. A redelivery can re-run alert eval safely: counting a failure and disabling a destination are both idempotent, so re-evaluating the same attempt changes nothing, and the decision never depends on delivery. The only thing that must not repeat is the opevent send, so we make delivery idempotent too: a send that already succeeded is skipped on replay, a send that failed is retried. On a redelivery, eval re-runs against the live count and we re-send only what wasn't delivered, so we never lose an alert and rarely duplicate one. All three opevent types (consecutive-failure, disabled, exhausted-retries) go through this one path. (Mechanics in Implementation notes below.)
When things back up. The in-memory queues are bounded. If delivery falls behind, it pushes back on Stage 1, then the batcher, then logmq stops pulling — so the backlog just waits in the broker. The existing DLQ catches messages that keep failing. Only a sink outage that affects everything can back up delivery, and even then we never lose data — we just retry and may send duplicates.
Health signal. Because the queues are in-memory, there's no separate health surface to watch. A delivery backlog backpressures into logmq, so it shows up as logmq consumer lag, the same signal operators already monitor. If logmq is healthy, the whole pipeline is.
Trade-off we're accepting: this keeps today's coupling — a total sink outage still backpressures saving (degrades to retries + DLQ, never data loss). Full isolation would need a durable opevent queue, which we're rejecting as overkill (see constraint 3).
Open questions (need input)
Sizing: derive from an existing scale knob instead of adding new ones. We're open source, so we can't pick one right number for everyone — but we also don't want a pile of new knobs. LOG_BATCH_SIZE (default 1000) already tells us the scale: it's how many entries land at once after each insert, i.e. the burst the eval/delivery pools have to drain before the next batch arrives. Proposal: size the eval/delivery pools and queues as a function of LOG_BATCH_SIZE, so scaling stays a single knob operators already set. Bigger batch → bigger pools, automatically, no redesign.
Design target: the system must be healthy at ~10k/s with one opevent per entry. Almost every real deployment will be far below that, but the design must not cap out there — it should scale beyond 10k/s when needed (10x, 100x) by raising the existing knob, without a redesign.
Eval workers: must keep up with saving; fast (~couple ms each), ordered per destination — so a small fraction of the batch covers it.
Delivery workers: cheap, order doesn't matter, and it's the slow part — so the larger share. Need to decide the ratio, the cap, and what backpressures when the queue is full.
Open question: the exact relationship to LOG_BATCH_SIZE (and whether LOG_BATCH_THRESHOLD_SECONDS factors in, since it bounds how often a batch arrives), and whether we still want a direct override for the rare operator who needs one.
logmq visibility timeout (~60s today). Leaning toward keeping it as-is — just flagging it so we confirm 60s still gives enough headroom for a message to finish eval + delivery under this design before the broker redelivers it.
Implementation notes (mechanics, for reviewers who want them)
How the idempotency above actually works, reusing what already exists:
Failure count stays a Redis set of attempt IDs per destination (SADD the attempt, SCARD for the count). SADD is idempotent, so re-counting an attempt on replay is a no-op. A success deletes the set (the reset).
Opevent dedup reuses the existing idempotence package (already used by the exhausted-retries alert). The delivery unit is the entry, so there's one key per log entry, covering all of that entry's opevents (1, 2, or 3) together. We reserve the key, send the entry's opevents, and mark the key processed only if all succeed; on any failure we clear the key so a later redelivery retries the whole entry. That gives "skip the entry if already delivered, retry it if not" for free. (A retry re-sends all of the entry's opevents, so the rare partial case — one of two already went out — produces a tolerated duplicate, not a loss.)
Two layers, don't conflate them:
Delivery layer — one key per entry, e.g. opevent:<attemptID> (the attempt ID is already destination-specific). Gates every opevent's delivery (consecutive-failure, disabled, exhausted alike): "did we already deliver this entry's opevents?" This is the only new key, and it's purely for logmq redelivery dedup.
Decision layer — exhausted-retries additionally has its own opevents:exhausted:<eventID>:<destinationID> key, unchanged. It's a suppression window (TTL = ExhaustedRetries.WindowSeconds) deciding whether exhausted fires at all for an event→destination in the window (e.g. so a later manual retry that exhausts again doesn't re-fire). It's exhausted-specific feature state, governed by its own config, out of scope for this RFC.
So exhausted-retries passes through both: the window decides whether to emit, the per-entry key dedups the delivery. The other opevent types only use the per-entry key.
Removed: the bespoke cfeval "evaluated" set and its MarkAttemptEvaluated path. The per-entry key replaces it, and it's opevent-delivery-scoped rather than alert-scoped, which is what it actually tracks.
Other designs we considered (A, B) — and why not
Two things had to be true at once: (1) keep per-destination order across batches, and (2) keep the slow delivery from slowing down saving.
design
(1) keeps order
(2) delivery doesn't slow saving
A — spawn workers per batch, wait for them
yes
no — the slowest destination holds up the batch
A′ — spawn per batch, don't wait
no — order goes random
yes
B — long-lived workers per destination, but send opevents in the same ordered step
yes
mostly, but see below
C (this proposal) — split eval (in order) from delivery (parallel)
yes
yes
Workers in A/A′ live for only one batch, so they don't remember what's still pending for a destination from the last batch. That forces a bad choice:
A — spawn per batch, wait for them. Order is kept, but the batcher waits for every worker before the next batch, so the slowest destination holds up saving.
flowchart LR
BP["batcher (1 goroutine)<br/>InsertMany (save batch)"] -->|"per batch: spawn 1 worker per destination"| S
subgraph S["ephemeral workers (this batch only)"]
WA["dest A<br/>eval + send (welded)"]
WB["dest B<br/>eval + send (welded)"]
end
WA --> ACK["ack / nack"]
WB --> ACK
ACK -.->|"next batch waits for all workers<br/>(slowest destination holds it up)"| NEXT(["next batch starts"])
Loading
A′ — spawn per batch, don't wait. The batcher unlocks right after spawning, so saving isn't held up. But with no memory across batches, per-destination order goes random.
flowchart LR
BP["batcher (1 goroutine)<br/>InsertMany (save batch)"] -->|"per batch: spawn 1 worker per destination"| S
subgraph S["ephemeral workers (this batch only)"]
WA["dest A<br/>eval + send (welded)"]
WB["dest B<br/>eval + send (welded)"]
end
WA --> ACK["ack / nack"]
WB --> ACK
BP -.->|"unlocks right after spawn<br/>(order across batches goes random)"| NEXT(["next batch starts"])
Loading
B — long-lived sharded workers, but send welded into the ordered lane. Same "same destination → same worker" as C, so order is kept and the batcher unlocks after dispatch. But the slow send sits inside the ordered shard worker: a destination that keeps failing fills its queue, messages sit too long, and the broker redelivers them, causing a retry storm.
flowchart LR
BP["batcher (1 goroutine)<br/>InsertMany (save batch)"] -->|"per entry, sharded by destination"| Q
BP -.->|"unlocks after dispatch"| NEXT(["next batch starts"])
subgraph S["N long-lived workers (ordered per destination)"]
Q["N queues<br/>sharded by destination"] --> W1["worker 1<br/>eval + send (welded)"]
Q --> WN["worker N<br/>eval + send (welded)"]
end
W1 --> ACK["ack / nack"]
WN --> ACK
Loading
Compare with the proposal: C keeps B's sharded ordered eval but forks the slow send out to one shared, unordered delivery pool. That fork is the whole difference.
How it works today (for reference)
flowchart LR
UP["deliverymq workers (parallel)"] -->|publish log entry| LQ[logmq] --> BP
subgraph S["batcher — 1 goroutine — single-threaded"]
BP["InsertMany (save batch)"] -->|"per entry, one at a time"| W["1 worker<br/>alert eval + send opevent (welded)"]
end
W --> ACK["ack / nack logmq msg"]
ACK -.->|"only now: next batch<br/>(batcher held through the whole loop)"| NEXT(["next batch starts"])
Loading
Because the last step runs one entry at a time, the whole pipeline can only go about as fast as 1 / per-entry time (~20/s at 50ms each), no matter how parallel everything upstream is. The slow sink send sits right inside that loop.
TL;DR
When logmq processes a batch, it first saves all the log entries in one batched write. That part is fine. The problem is what happens next: for each entry, it runs alert evaluation and then sends an opevent, one entry at a time, in a single goroutine. The slowest step is sending the opevent to the customer's sink, so one slow sink slows the whole loop down, including saving new logs.
Proposal: take that post-save work off the saving path so it can't block saving, and run it in parallel instead of one at a time. Split it into two parts:
Both run in memory. We don't add any new queue or infra. We keep the logmq message un-acked until delivery finishes, so nothing is lost.
What we need to get right (please push back here)
If you disagree with one of these, the design changes.
1. Keep per-destination order in alert eval — roughly, not perfectly. (Order doesn't matter for opevent delivery.)
Why order matters for alert eval, and why we only aim for "roughly"
Alerting counts how many times a destination failed in a row. A
successresets that count. So if an oldsuccessgets processed in the middle of a run of failures, the count resets at the wrong time and the alert (or auto-disable) can fire late, or get missed. So the order we process attempts in does matter.But we can't promise perfect order, and we don't today. The order is already a bit loose before logmq even sees it:
publishmq → deliverymq → logmquses parallel workers and retries, so attempts for one destination can show up slightly out of order. Whatever order logmq receives, that's what it works with.So the goal is: stay close to the real order per destination, like it is today — not perfect. Today everything runs one at a time, so it's roughly in order. The thing we must avoid: if we just run everything fully in parallel, the order per destination becomes random. That's the part we want to prevent. The fix is to always send attempts for the same destination to the same worker, so they stay in order. Different destinations run fully in parallel — their order relative to each other never mattered.
2. A slow opevent delivery must not slow down saving logs.
Where the slow part actually is
Alert evaluation is fast — roughly a couple of milliseconds, just a couple of Redis calls (count the failures, check the threshold, disable the destination if needed).
The slow part is sending the opevent to the sink — tens of milliseconds normally. We can set an aggressive timeout (say a few seconds) to cap the worst case, but done serially even that becomes a huge bottleneck: if a sink has an issue, every send waits out the timeout, one after another. Today that send happens inside the same one-at-a-time loop that the pipeline waits on. So when one customer's sink is slow, the loop slows down, the batch processor slows down, logmq stops pulling new messages, and saving logs slows down too. One slow sink drags down the whole pipeline. It can also push messages past their visibility timeout while they wait, so the broker redelivers them and we reprocess the same work.
Note: alert evaluation still has to keep up with how fast we save logs, but that's easy — it's fast, so we just run a few of them at once.
3. No new durable queue (no
opeventsmq).Why we rely on the logmq message instead
Sending an opevent should work most of the time. For the rare failure, we don't ack the logmq message — it gets redelivered and we redo the work. That's safe because the whole pipeline is idempotent: re-inserting the log doesn't create a duplicate, and re-running alert eval doesn't double-count or double-emit. So we can nack freely. Adding a separate durable queue just for opevents isn't worth the extra complexity. The in-memory parts below are only for running things in parallel, not for durability.
Proposal
After the batched save, dispatch each entry into two in-memory stages instead of the one-at-a-time loop. The message stays un-acked until delivery finishes (ack on success, nack to retry on failure). The two stages have different shapes: Stage 1 is sharded (keeps order per destination), Stage 2 is shared (parallelism).
flowchart LR BP["batcher (1 goroutine)<br/>InsertMany (save batch)"] -->|"per entry, routed by hash(destID)"| Q1 BP -.->|"unlocks after dispatch,<br/>doesn't wait for eval/delivery"| NEXT(["next batch starts"]) subgraph S1["Stage 1 — opevent eval (ordered per destination)"] Q1["N queues<br/>sharded by destination"] --> W1["eval worker 1"] Q1 --> WN["eval worker N"] end W1 -->|opevents| Q2 WN -->|opevents| Q2 subgraph S2["Stage 2 — delivery (unordered)"] Q2["1 shared queue"] --> D1["delivery worker 1"] Q2 --> DM["delivery worker M"] end D1 --> ACK["ack / nack logmq msg"] DM --> ACKDesign assumption: one opevent per log entry is the expected case, not a rare spike — we intend to support opevents generated per entry. So we size and reason as if each entry generates one; we don't lean on "most entries emit nothing."
Trade-off we're accepting: this keeps today's coupling — a total sink outage still backpressures saving (degrades to retries + DLQ, never data loss). Full isolation would need a durable opevent queue, which we're rejecting as overkill (see constraint 3).
Open questions (need input)
LOG_BATCH_SIZE(default 1000) already tells us the scale: it's how many entries land at once after each insert, i.e. the burst the eval/delivery pools have to drain before the next batch arrives. Proposal: size the eval/delivery pools and queues as a function ofLOG_BATCH_SIZE, so scaling stays a single knob operators already set. Bigger batch → bigger pools, automatically, no redesign.LOG_BATCH_SIZE(and whetherLOG_BATCH_THRESHOLD_SECONDSfactors in, since it bounds how often a batch arrives), and whether we still want a direct override for the rare operator who needs one.Implementation notes (mechanics, for reviewers who want them)
How the idempotency above actually works, reusing what already exists:
SADDthe attempt,SCARDfor the count).SADDis idempotent, so re-counting an attempt on replay is a no-op. Asuccessdeletes the set (the reset).idempotencepackage (already used by the exhausted-retries alert). The delivery unit is the entry, so there's one key per log entry, covering all of that entry's opevents (1, 2, or 3) together. We reserve the key, send the entry's opevents, and mark the key processed only if all succeed; on any failure we clear the key so a later redelivery retries the whole entry. That gives "skip the entry if already delivered, retry it if not" for free. (A retry re-sends all of the entry's opevents, so the rare partial case — one of two already went out — produces a tolerated duplicate, not a loss.)opevent:<attemptID>(the attempt ID is already destination-specific). Gates every opevent's delivery (consecutive-failure, disabled, exhausted alike): "did we already deliver this entry's opevents?" This is the only new key, and it's purely for logmq redelivery dedup.opevents:exhausted:<eventID>:<destinationID>key, unchanged. It's a suppression window (TTL =ExhaustedRetries.WindowSeconds) deciding whether exhausted fires at all for an event→destination in the window (e.g. so a later manual retry that exhausts again doesn't re-fire). It's exhausted-specific feature state, governed by its own config, out of scope for this RFC.cfeval"evaluated" set and itsMarkAttemptEvaluatedpath. The per-entry key replaces it, and it's opevent-delivery-scoped rather than alert-scoped, which is what it actually tracks.Other designs we considered (A, B) — and why not
Two things had to be true at once: (1) keep per-destination order across batches, and (2) keep the slow delivery from slowing down saving.
Workers in A/A′ live for only one batch, so they don't remember what's still pending for a destination from the last batch. That forces a bad choice:
A — spawn per batch, wait for them. Order is kept, but the batcher waits for every worker before the next batch, so the slowest destination holds up saving.
flowchart LR BP["batcher (1 goroutine)<br/>InsertMany (save batch)"] -->|"per batch: spawn 1 worker per destination"| S subgraph S["ephemeral workers (this batch only)"] WA["dest A<br/>eval + send (welded)"] WB["dest B<br/>eval + send (welded)"] end WA --> ACK["ack / nack"] WB --> ACK ACK -.->|"next batch waits for all workers<br/>(slowest destination holds it up)"| NEXT(["next batch starts"])A′ — spawn per batch, don't wait. The batcher unlocks right after spawning, so saving isn't held up. But with no memory across batches, per-destination order goes random.
flowchart LR BP["batcher (1 goroutine)<br/>InsertMany (save batch)"] -->|"per batch: spawn 1 worker per destination"| S subgraph S["ephemeral workers (this batch only)"] WA["dest A<br/>eval + send (welded)"] WB["dest B<br/>eval + send (welded)"] end WA --> ACK["ack / nack"] WB --> ACK BP -.->|"unlocks right after spawn<br/>(order across batches goes random)"| NEXT(["next batch starts"])B — long-lived sharded workers, but send welded into the ordered lane. Same "same destination → same worker" as C, so order is kept and the batcher unlocks after dispatch. But the slow send sits inside the ordered shard worker: a destination that keeps failing fills its queue, messages sit too long, and the broker redelivers them, causing a retry storm.
flowchart LR BP["batcher (1 goroutine)<br/>InsertMany (save batch)"] -->|"per entry, sharded by destination"| Q BP -.->|"unlocks after dispatch"| NEXT(["next batch starts"]) subgraph S["N long-lived workers (ordered per destination)"] Q["N queues<br/>sharded by destination"] --> W1["worker 1<br/>eval + send (welded)"] Q --> WN["worker N<br/>eval + send (welded)"] end W1 --> ACK["ack / nack"] WN --> ACKCompare with the proposal: C keeps B's sharded ordered eval but forks the slow send out to one shared, unordered delivery pool. That fork is the whole difference.
How it works today (for reference)
flowchart LR UP["deliverymq workers (parallel)"] -->|publish log entry| LQ[logmq] --> BP subgraph S["batcher — 1 goroutine — single-threaded"] BP["InsertMany (save batch)"] -->|"per entry, one at a time"| W["1 worker<br/>alert eval + send opevent (welded)"] end W --> ACK["ack / nack logmq msg"] ACK -.->|"only now: next batch<br/>(batcher held through the whole loop)"| NEXT(["next batch starts"])Because the last step runs one entry at a time, the whole pipeline can only go about as fast as 1 / per-entry time (~20/s at 50ms each), no matter how parallel everything upstream is. The slow sink send sits right inside that loop.