broker: the substrate pre-declares a consumer's dead-lettered queue (ADR 0048)
LavinMQ refuses a non-administrator declaring a queue with a dead-letter exchange, so a scoped module cannot make its own. EnsureModuleQueue declares <node>.<module>.events with its DLX as the mesh, and 'module issue' does so for a consuming module — the runtime then passively checks it rather than declaring. Verified against a real broker: the scoped account binds and consumes the pre-declared queue. Claude-Session: https://claude.ai/code/session_01LrgweAeERJYBg88c5cKDzF
This commit is contained in:
@@ -243,6 +243,13 @@ func moduleCommand(ctx context.Context, args []string) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
// A consumer's queue, with its dead-letter, is the substrate's to declare — its own account
|
||||||
|
// may not (ADR 0048). Made now, so it exists before the module binds onto it.
|
||||||
|
if len(m.Consumes) > 0 {
|
||||||
|
if err := management.EnsureModuleQueue(ctx, *forNode, module); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
known, err := broker.FromEnvironment()
|
known, err := broker.FromEnvironment()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -143,6 +143,21 @@ func (m *Management) CreateModuleAccount(ctx context.Context, node, module, pass
|
|||||||
return account, nil
|
return account, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// EnsureModuleQueue declares a consuming module's queue with its dead-letter exchange, idempotently.
|
||||||
|
// The substrate declares it because a scoped module account may not: the broker refuses a queue with
|
||||||
|
// a dead-letter exchange to a non-administrator (novox/hq ADR 0048), so a consumer passively checks
|
||||||
|
// the queue the mesh made rather than declaring its own.
|
||||||
|
func (m *Management) EnsureModuleQueue(ctx context.Context, node, module string) error {
|
||||||
|
queue := ModuleQueueFor(node, module)
|
||||||
|
if err := m.put(ctx, "/api/queues/%2f/"+url.PathEscape(queue), map[string]any{
|
||||||
|
"durable": true,
|
||||||
|
"arguments": map[string]any{"x-dead-letter-exchange": DeadExchangeName},
|
||||||
|
}); err != nil {
|
||||||
|
return fmt.Errorf("cannot declare the queue for %s on %s: %w", module, node, err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// EnsureEventExchanges declares the bus's exchanges and the dead-letter home, idempotently. The
|
// EnsureEventExchanges declares the bus's exchanges and the dead-letter home, idempotently. The
|
||||||
// substrate owns them (a module's account may not declare an exchange), and a dead-letter exchange
|
// substrate owns them (a module's account may not declare an exchange), and a dead-letter exchange
|
||||||
// with no queue behind it drops what it receives — so a durable queue bound to `#` retains a poison
|
// with no queue behind it drops what it receives — so a durable queue bound to `#` retains a poison
|
||||||
|
|||||||
Reference in New Issue
Block a user