mesh-controller.try fault A: decide drop-only dead letters by a module's own retention: work-queue, not by the bus's work-queue retention alone (issue 1178)
sourceAge keyed AgainTo's clause-6 refusal on the stream's bus retention policy, which SEAT_* and other mesh-internal work queues share without being one of ADR 0436 §6's module streams — so every seat's ask, once given up on, was refused delivery again. Added a dedicated metadata key (WorkQueueMetadata), set only by ModuleStream.Stream() when the catalogue declared retention: work-queue, and read sourceAge from that instead of the stream's retention policy. New test proves a seat's own work queue carries neither that metadata nor the restricted one.
This commit is contained in:
@@ -100,8 +100,16 @@ const withheldWords = "its body and its own headers are not shown: it came from
|
||||
"the modules its grant names may read (novox/hq ADR 0376)"
|
||||
|
||||
// sourceAge is whether a stream is restricted (novox/hq ADR 0363 §4; a work-queue stream is restricted too,
|
||||
// novox/hq ADR 0436 §6, aged and hidden the same way), whether it is a work queue, and its maximum age: read
|
||||
// from the stream itself, where the controller says all three. A stream that no longer exists is neither.
|
||||
// novox/hq ADR 0436 §6, aged and hidden the same way), whether it is a module stream declared retention:
|
||||
// work-queue, and its maximum age: read from the stream itself, where the controller says all three. A
|
||||
// stream that no longer exists is neither.
|
||||
//
|
||||
// **The module declaration, never the bus's work-queue retention alone** (novox/hq ADR 0436 §6, found by
|
||||
// mesh-controller.try on d8249dad): SEAT_* and other mesh-internal streams are JetStream work queues too —
|
||||
// exactly one worker takes each ask (ADR 0264 §4) — and their dead letters are still delivered again. Only
|
||||
// a module stream the catalogue declared retention: work-queue says so in its own metadata
|
||||
// (broker.WorkQueueMetadata, set by ModuleStream.Stream() alone); checking the bus's retention policy
|
||||
// caught every seat's queue too and refused every ask ever given up on delivered again.
|
||||
func sourceAge(js nats.JetStreamContext, stream string) (restricted, workQueue bool, maxAge time.Duration, err error) {
|
||||
info, err := js.StreamInfo(stream)
|
||||
if errors.Is(err, nats.ErrStreamNotFound) {
|
||||
@@ -110,8 +118,8 @@ func sourceAge(js nats.JetStreamContext, stream string) (restricted, workQueue b
|
||||
if err != nil {
|
||||
return false, false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err)
|
||||
}
|
||||
workQueue = info.Config.Retention == nats.WorkQueuePolicy
|
||||
return info.Config.Metadata[broker.RestrictedMetadata] == "true", workQueue, info.Config.MaxAge, nil
|
||||
return info.Config.Metadata[broker.RestrictedMetadata] == "true",
|
||||
info.Config.Metadata[broker.WorkQueueMetadata] == "true", info.Config.MaxAge, nil
|
||||
}
|
||||
|
||||
// ttlOf is a time left as a message's time to live, in whole seconds and never more than what is left.
|
||||
|
||||
Reference in New Issue
Block a user