diff --git a/modules/mesh-delivery-view/internal/view/backlog.go b/modules/mesh-delivery-view/internal/view/backlog.go index 1a3f17e9..3fe7b11b 100644 --- a/modules/mesh-delivery-view/internal/view/backlog.go +++ b/modules/mesh-delivery-view/internal/view/backlog.go @@ -2,6 +2,7 @@ package view import ( "context" + "errors" "fmt" "time" @@ -55,13 +56,32 @@ func (b *Backlog) Rows() []BacklogRow { } row.Waiting, row.Held = waiting, held if waiting > 0 { - body, err := b.Message(stream, subject, next) - if err == nil { - var a buildstep.StepAsk - a, err = buildstep.DecodeStepAsk(body) - if err == nil { - row.Oldest, row.OldestSecs = a.At, int64(now().Sub(a.At).Seconds()) + // Both forms of the accept subject: main's and the asker's (design 47 G §600, §484); the oldest of the two. + var oldest *buildstep.StepAsk + var err error + for _, form := range []string{subject, subject + ".*"} { + body, gerr := b.Message(stream, form, next) + if errors.Is(gerr, jetstream.ErrMsgNotFound) { + continue } + if gerr != nil { + err = gerr + break + } + a, derr := buildstep.DecodeStepAsk(body) + if derr != nil { + err = derr + break + } + if oldest == nil || a.At.Before(oldest.At) { + oldest = &a + } + } + if err == nil && oldest == nil { + err = jetstream.ErrMsgNotFound + } + if err == nil { + row.Oldest, row.OldestSecs = oldest.At, int64(now().Sub(oldest.At).Seconds()) } if err != nil { row.Error = fmt.Sprintf("the oldest waiting ask cannot be read: %v", err) diff --git a/modules/mesh-delivery-view/internal/view/backlog_test.go b/modules/mesh-delivery-view/internal/view/backlog_test.go index e8dc2fa9..4e704170 100644 --- a/modules/mesh-delivery-view/internal/view/backlog_test.go +++ b/modules/mesh-delivery-view/internal/view/backlog_test.go @@ -9,6 +9,7 @@ import ( "time" "git.novox.be/novox/mesh-delivery/sdk/buildstep" + "github.com/nats-io/nats.go/jetstream" ) func stepAsk(t *testing.T, id, kind string, at time.Time) []byte { @@ -37,10 +38,16 @@ func fakeSeats(t *testing.T, now time.Time) *Backlog { return 0, 0, 1, nil }, Message: func(stream, s string, seq uint64) ([]byte, error) { - if s != subject || seq != 42 { + if seq != 42 { return nil, fmt.Errorf("asked %s at %d", s, seq) } - return oldest, nil + switch s { + case subject: // main's form holds a younger ask + return stepAsk(t, "m0000002bbbbbb", buildstep.AskCheck, now.Add(-30*time.Second)), nil + case subject + ".*": // the asker's form holds the oldest + return oldest, nil + } + return nil, jetstream.ErrMsgNotFound }} }