grants part, indexer and checker read the seat list again through the sdk's ReloadIfUnread; the checker's hub binds a late consumer (B9e; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -461,12 +461,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return nil
|
||||
}
|
||||
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "checker_first_views reads nothing and checker_first_view_again is not heard", logger.Printf))
|
||||
// Every part has registered. Where the consumer is not there yet no package registry's published is heard and a group's
|
||||
// check fails at its publish bound until the checker starts again.
|
||||
if err := hub.Start(ctx, js, cred); err != nil {
|
||||
logger.Printf("%v, so no package registry's published is heard and a group's check fails at its publish bound until the checker starts again", err)
|
||||
} else {
|
||||
readIfUnread(seats.reader)
|
||||
// Every part has registered. Where the consumer is not there yet the hub binds it as soon as it is, and the seat list is read
|
||||
// again then if it is still unread.
|
||||
if err := hub.StartUntilBound(ctx, js, cred, nil, func() { readIfUnread(seats.reader) }); err != nil {
|
||||
logger.Printf("%v, so no package registry's published is heard until the consumer is there; the checker tries again", err)
|
||||
}
|
||||
defer hub.Stop()
|
||||
// The checks, the tries and the hand asks, once every first view is answered (the lead's ruling for DLV-M2): a check
|
||||
|
||||
@@ -54,8 +54,7 @@ func (s seatList) Definitions() ([]json.RawMessage, error) {
|
||||
// checker's consumer existed, whose events start at the stream's end: the first read found it silent and nothing else reads
|
||||
// again until its next event, so while the list is unread it is read now, on a goroutine of its own (ADR 0636 §1).
|
||||
func readIfUnread(reader *seatlist.Reader) {
|
||||
if reader == nil || reader.Status().Heard {
|
||||
return
|
||||
if reader != nil {
|
||||
reader.ReloadIfUnread()
|
||||
}
|
||||
go func() { _ = reader.Reload() }()
|
||||
}
|
||||
|
||||
@@ -142,9 +142,7 @@ func readWhenBound(reader *seatlist.Reader, states ...bucket) {
|
||||
}
|
||||
for _, state := range states {
|
||||
state.Ready(func(jetstream.KeyValue) {
|
||||
if !reader.Status().Heard {
|
||||
go func() { _ = reader.Reload() }()
|
||||
}
|
||||
reader.ReloadIfUnread()
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -54,11 +54,9 @@ func (g *seatGate) follow(bus seatlist.Bus, changed func(), logf func(string, ..
|
||||
// part reads it now (ADR 0636 §1), on a goroutine of its own so the pass is not held; the reader changes the catalogue and wakes
|
||||
// the part when the seat answers, and stays unread, to be read at the seat's next event, when it does not.
|
||||
func (g *seatGate) eventsLive() {
|
||||
r := g.reader.Load()
|
||||
if r == nil || r.Status().Heard {
|
||||
return
|
||||
if r := g.reader.Load(); r != nil {
|
||||
r.ReloadIfUnread()
|
||||
}
|
||||
go func() { _ = r.Reload() }()
|
||||
}
|
||||
|
||||
// tools are the two tools over whatever reader the gate holds at the call.
|
||||
|
||||
Reference in New Issue
Block a user