diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 836675d4..fa04d0b8 100644 --- a/internal/broker/jetstream.go +++ b/internal/broker/jetstream.go @@ -168,8 +168,14 @@ func StreamConfigFor(s Stream) *nats.StreamConfig { } want.AllowDirect = s.Direct want.AllowMsgTTL = s.AllowMsgTTL - if s.Restricted { - want.Metadata = map[string]string{RestrictedMetadata: "true"} + if s.Restricted || s.WorkQueue { + want.Metadata = map[string]string{} + if s.Restricted { + want.Metadata[RestrictedMetadata] = "true" + } + if s.WorkQueue { + want.Metadata[WorkQueueMetadata] = "true" + } } if s.MaxBytes > 0 { want.MaxBytes = s.MaxBytes diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index 4a79df6d..3dc94a30 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -98,6 +98,10 @@ func (s ModuleStream) Stream() Stream { // by (novox/hq ADR 0436 §6, ADR 0376 §3-4): a work-queue stream's grant is always empty (ADR 0436 // §6 refuses one that carries a grant), so this never collides with a trusted-config restriction. out.Restricted = true + // And its own metadata, apart from the bus's work-queue retention SEAT_* and other mesh-internal + // streams share without being one of this clause's: only this flag says a dead letter kept from the + // stream is drop-only (novox/hq ADR 0436 §6), never the stream's Retention alone. + out.WorkQueue = true } return out } diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 8be30cbb..d297d917 100644 --- a/internal/broker/streams.go +++ b/internal/broker/streams.go @@ -65,11 +65,21 @@ type Stream struct { // Restricted marks a module's stream restricted by a grant (novox/hq ADR 0363 §4), in the stream's own // metadata, so whoever keeps one of its messages reads it from the bus (ADR 0376 §3). Restricted bool + // WorkQueue marks a module stream declared retention: work-queue (novox/hq ADR 0436 §6), in the + // stream's own metadata, so a dead letter kept from it is read as drop-only from the bus, by the + // declaration and not by the bus's work-queue retention alone — which SEAT_* and other mesh-internal + // streams share without being one of ADR 0436's, and whose dead letters are still delivered again + // (ADR 0264 §4). + WorkQueue bool } // RestrictedMetadata is the key a restricted stream's metadata says it under. const RestrictedMetadata = "mesh-restricted" +// WorkQueueMetadata is the key a module stream declared retention: work-queue (novox/hq ADR 0436 §6) says +// it under: never set by a mesh-internal work-queue stream (a seat's, a check's), only by Stream.WorkQueue. +const WorkQueueMetadata = "mesh-work-queue" + // AssignmentsStream holds every assignment's membership, the newest per subject. const AssignmentsStream = "ASSIGNMENTS" diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index b806ad2d..94de7d22 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -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. diff --git a/internal/link/deadletters_work_queue_test.go b/internal/link/deadletters_work_queue_test.go index 8aa5b74b..2a8d6946 100644 --- a/internal/link/deadletters_work_queue_test.go +++ b/internal/link/deadletters_work_queue_test.go @@ -31,6 +31,25 @@ func TestAWorkQueueStreamIsRestrictedOnTheBus(t *testing.T) { } } +// mesh-controller.try fault A, d8249dad: a seat's own work queue (SEAT_*, novox/hq ADR 0190, ADR 0264 §4) is +// a JetStream work queue at the bus level too — exactly one worker takes each ask — but it is never a module +// stream the catalogue declared retention: work-queue (ADR 0436 §6), so it carries none of that clause's +// metadata and its dead letters are still delivered again (ADR 0264 §4), unlike mail-sends'. +func TestASeatsWorkQueueIsNotADR0436WorkQueue(t *testing.T) { + seatQueue := broker.Stream{Name: "SEAT_NODE_BUILD_AGENT", Retention: broker.RetentionWorkQueue, DiscardNew: true} + cfg := broker.StreamConfigFor(seatQueue) + if _, has := cfg.Metadata[broker.WorkQueueMetadata]; has { + t.Fatalf("a seat's own work queue carries ADR 0436 §6's metadata: %+v", cfg.Metadata) + } + if _, has := cfg.Metadata[broker.RestrictedMetadata]; has { + t.Fatalf("a seat's own work queue is restricted: %+v", cfg.Metadata) + } + // sourceAge (deadletters.go) reads exactly these two keys off a real stream's config; neither set means + // KeepDeadLetter marks a seat's dead letter WorkQueue: false, so AgainTo never reaches clause 6's + // drop-only case for it (proven directly in TestAWorkQueueDeadLetterIsNeverDeliveredAgain) and a seat's + // ask is still delivered again (ADR 0264 §4), as it was before this release. +} + // 1178/AC3: a dead letter's headers say it came from a work-queue stream, and reading it back carries that // (and the restricted flag its aging and hiding already key off, deadLetterOf being the one place both are // read: see DeadLetter.Restricted's doc).