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:
@@ -168,8 +168,14 @@ func StreamConfigFor(s Stream) *nats.StreamConfig {
|
|||||||
}
|
}
|
||||||
want.AllowDirect = s.Direct
|
want.AllowDirect = s.Direct
|
||||||
want.AllowMsgTTL = s.AllowMsgTTL
|
want.AllowMsgTTL = s.AllowMsgTTL
|
||||||
|
if s.Restricted || s.WorkQueue {
|
||||||
|
want.Metadata = map[string]string{}
|
||||||
if s.Restricted {
|
if s.Restricted {
|
||||||
want.Metadata = map[string]string{RestrictedMetadata: "true"}
|
want.Metadata[RestrictedMetadata] = "true"
|
||||||
|
}
|
||||||
|
if s.WorkQueue {
|
||||||
|
want.Metadata[WorkQueueMetadata] = "true"
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if s.MaxBytes > 0 {
|
if s.MaxBytes > 0 {
|
||||||
want.MaxBytes = s.MaxBytes
|
want.MaxBytes = s.MaxBytes
|
||||||
|
|||||||
@@ -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
|
// 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.
|
// §6 refuses one that carries a grant), so this never collides with a trusted-config restriction.
|
||||||
out.Restricted = true
|
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
|
return out
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
// 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).
|
// metadata, so whoever keeps one of its messages reads it from the bus (ADR 0376 §3).
|
||||||
Restricted bool
|
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.
|
// RestrictedMetadata is the key a restricted stream's metadata says it under.
|
||||||
const RestrictedMetadata = "mesh-restricted"
|
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.
|
// AssignmentsStream holds every assignment's membership, the newest per subject.
|
||||||
const AssignmentsStream = "ASSIGNMENTS"
|
const AssignmentsStream = "ASSIGNMENTS"
|
||||||
|
|
||||||
|
|||||||
@@ -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)"
|
"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,
|
// 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
|
// novox/hq ADR 0436 §6, aged and hidden the same way), whether it is a module stream declared retention:
|
||||||
// from the stream itself, where the controller says all three. A stream that no longer exists is neither.
|
// 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) {
|
func sourceAge(js nats.JetStreamContext, stream string) (restricted, workQueue bool, maxAge time.Duration, err error) {
|
||||||
info, err := js.StreamInfo(stream)
|
info, err := js.StreamInfo(stream)
|
||||||
if errors.Is(err, nats.ErrStreamNotFound) {
|
if errors.Is(err, nats.ErrStreamNotFound) {
|
||||||
@@ -110,8 +118,8 @@ func sourceAge(js nats.JetStreamContext, stream string) (restricted, workQueue b
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return false, false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err)
|
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",
|
||||||
return info.Config.Metadata[broker.RestrictedMetadata] == "true", workQueue, info.Config.MaxAge, nil
|
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.
|
// ttlOf is a time left as a message's time to live, in whole seconds and never more than what is left.
|
||||||
|
|||||||
@@ -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
|
// 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
|
// (and the restricted flag its aging and hiding already key off, deadLetterOf being the one place both are
|
||||||
// read: see DeadLetter.Restricted's doc).
|
// read: see DeadLetter.Restricted's doc).
|
||||||
|
|||||||
Reference in New Issue
Block a user