Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

Memory-bound concurrency for queue consumers

Aperta
  • Qualcosa non va?
#1,096 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
5/5
Tempo stimato
Più di una settimana
Idoneità per principianti
30/100
Tipo di issue
Funzionalità
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
aws, azure, go, google-cloud, rabbitmq
Ambito
distributed-systems, infrastructure

Direzione di ricerca

Start by reading the queue-consumer flow, especially where messages are received before waiting for a concurrency slot. Compare the existing GCP Pub/Sub ReceiveSettings.MaxOutstandingMessages setup with its byte flow-control setting; also inspect how SQS, Azure Service Bus, and RabbitMQ handle pulling or prefetching. Done means each queue can enforce the agreed byte limit alongside concurrency, including the oversized-message case, with tests covering those behaviors.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

enhancement internalmq

Outpost limits how much work each internal queue consumer holds with a message count: PUBLISH_MAX_CONCURRENCY, DELIVERY_MAX_CONCURRENCY and LOG_MAX_CONCURRENCY. Memory, though, follows count × message size, and nothing bounds the size side.

The same concurrency setting is safe or fatal depending on payload size. A value tuned for throughput with ~1 KB events keeps a few MB in flight. With ~1 MB events it keeps GBs in flight. Each body also exists several times in memory while it's handled: the decoded queue message, the event struct, the destination's request, and the attempt log (the event plus the response).

When it triggers

You need all three:

  • concurrency raised above the default of 1 for throughput
  • large payloads
  • a slow or throttling destination

Slow deliveries keep every slot full. Retries add to new traffic. The process then holds concurrency × payload size × copies of live heap. That memory is live data, so GOMEMLIMIT can't reclaim it, and the process is OOM-killed.

After a restart, the unacked messages are redelivered to a process with every slot free. It fills again and gets killed again, and the loop runs until concurrency is lowered by hand.

Today, avoiding this means lowering concurrency to fit the largest payloads. That caps throughput for everything else. If a deployment's traffic mixes small and large events, no single value is right.

Proposal

Add a byte-based limit per queue next to the existing concurrency setting, for example:

  • PUBLISH_MAX_CONCURRENCY_BYTES
  • DELIVERY_MAX_CONCURRENCY_BYTES
  • LOG_MAX_CONCURRENCY_BYTES

How the two settings work together:

  • Concurrency stays the throughput setting: how many messages are handled at once.
  • Concurrency bytes is the memory setting: how many bytes of messages can be in flight at once, measured by message body size.
  • A message starts only if both limits have room. When either is full, the consumer stops taking messages, so excess load waits in the broker as backlog instead of in process memory.
  • One message larger than the whole byte limit still starts when nothing else is in flight, so an oversized event can't stall the queue.

With a byte limit in place, operators can set concurrency high for throughput without having to size it for their largest payloads. Small events run at full concurrency; large events limit themselves.

Open questions

  • Default. Unset (no behavior change), or derived from GOMEMLIMIT so deployments get a safe bound without tuning?
  • Scope. One limit per queue, matching the existing concurrency settings, or one limit shared across the process, which maps more directly to the memory limit?

Implementation notes

Starting points only; the PR decides the approach.

  • GCP Pub/Sub has native byte flow control (ReceiveSettings.MaxOutstandingBytes). Today we set only MaxOutstandingMessages and leave bytes at the SDK default (1 GB).
  • SQS, Azure Service Bus and RabbitMQ have no native byte limit, so the consumer would enforce it. The consumer currently receives a message before waiting for a concurrency slot. The byte limit should apply before messages are pulled.
  • RabbitMQ subscriptions don't set a prefetch count, so the broker may push more messages than the concurrency setting allows.
Lingua principale
Go
Stelle
1.1k
Fork
70
Merge medio
1g 18h
PR unite (30g)
56

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di hookdeck/outpost

Tutte le issue di hookdeck/outpost

Issue simili

Altre issue su Go

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.