§484: the status view reads the oldest waiting ask in both forms of the accept subject
Builds: design 47 G §484, §600 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user