Repository navigation
Group-commit durable publishing + dedicated 'msgflux' schema - #2
Merged
Merged
Conversation
added 8 commits
June 23, 2026 10:27
… window DurableBuffer previously acked the publisher as soon as a message was buffered in memory, while the actual flush to the store happened only every interval or every N messages. A crash (or an unordered shutdown vs. the store) in that window silently dropped messages already reported as published — violating at-least-once. Switch to a group-commit model: AddAsync only completes once the batch containing its messages is durably persisted. Concurrent publishes are coalesced into a single batch (one flush in flight at a time), so batching throughput is preserved without a loss window. On persist failure the error surfaces to the caller (no silent internal retry). A bounded number of in-flight messages (MaxBufferedMessages, default 1000) applies backpressure instead of buffering unbounded. The final flush now runs via IHostedService during the host's ordered shutdown, while the store is still alive.
Three end-to-end integration tests over Testcontainers PostgreSQL: - a buffered publish is acknowledged only after its row is committed (group-commit); - a batched burst is persisted and each message is consumed exactly once; - messages still in the buffer at shutdown are flushed, not silently dropped (the scenario that motivated the change).
The inbox table now lives at msgflux.messages instead of public.msgflux_messages: the schema is created (CREATE SCHEMA IF NOT EXISTS msgflux) and every query is schema-qualified. Indexes live in the schema with the table; the COPY temp table (_msgflux_bulk) and the purge advisory-lock key are unaffected. Also realigned the unused schema.sql, which had drifted from the real DDL (message_id was TEXT instead of UUID), to the source-of-truth schema. BREAKING: existing deployments must migrate. For an existing table: CREATE SCHEMA IF NOT EXISTS msgflux; ALTER TABLE public.msgflux_messages SET SCHEMA msgflux; ALTER TABLE msgflux.msgflux_messages RENAME TO messages;
The DDL is owned by SchemaInitializer.SchemaSql; this file was never read at build or runtime (no embedded resource, no file read) and only drifted from the real schema. Deleting it removes a misleading duplicate.
- replace the 'No WAL / buffered messages are lost' trade-off with the group-commit guarantee (durable publish acked only after the batch is committed) and its real trade-off (bounded backpressure); - note group-commit + coalescing in the durable-delivery flow; - document the new MaxBufferedMessages option; - drop the removed PollingBatchSize from the options table and example.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Résumé
Corrige une fenêtre de perte at-least-once dans le
DurableBufferet déplace la table durable dans un schéma Postgres dédié.1. Group-commit durable publishing (fenêtre de perte fermée)
Auparavant, un publish at-least-once était acquitté dès la mise en buffer mémoire, alors que le flush vers le store n'avait lieu que par intervalle/seuil. Un crash (ou un shutdown non ordonné vis-à-vis du store) dans cette fenêtre perdait silencieusement des messages déjà signalés comme publiés.
AddAsyncne se complète qu'après la persistance effective du batch (group-commit).MaxBufferedMessages(défaut 1000). Au-delà, le publish attend → mémoire bornée, plus de risque d'OOM (le buffer était non borné).DurableBufferdevientIHostedService, son flush final s'exécute pendant l'arrêt du host, tant que le store est vivant.2. Schéma Postgres dédié
msgfluxLa table passe de
public.msgflux_messagesàmsgflux.messages(schéma créé viaCREATE SCHEMA IF NOT EXISTS, requêtes qualifiées). Suppression duschema.sqlmort/divergent.Tests
API : ajout de
WithMaxBufferedMessages(int); suppression déjà actée deWithPollingBatchSizereflétée dans la doc.