buildrun: open a steps watch again that closed having heard nothing, up to 3 closes in a row
A watch that closed at once was left for good, so an ask whose own watch went deaf was read only at a reconnect (review of fix/lost-check).
This commit is contained in:
@@ -175,8 +175,7 @@ func (s JetStreamAsks) AskSteps(ctx context.Context, module, askID string) ([]de
|
||||
|
||||
// HearEnds is the HeardEnds every asker of the build seats waits through: over inner (the asks' own watches, or a
|
||||
// LostAskWaiter's inner), reading with asks, and nudged at each step end of each of modules (one watch of each steps state,
|
||||
// updates only; one that closes is opened again while it heard something since it was opened, and said and left
|
||||
// otherwise). The asker nudges it at a reconnect too (Nudge("")).
|
||||
// updates only; one that closes is opened again, and said and left after reopenTries closes in a row that heard nothing). The asker nudges it at a reconnect too (Nudge("")).
|
||||
func HearEnds(ctx context.Context, asks JetStreamAsks, inner AskWaiter, modules []string, logf func(string, ...any)) *HeardEnds {
|
||||
if logf == nil {
|
||||
logf = func(string, ...any) {}
|
||||
@@ -194,6 +193,7 @@ func HearEnds(ctx context.Context, asks JetStreamAsks, inner AskWaiter, modules
|
||||
}
|
||||
|
||||
func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module string) {
|
||||
silent := 0 // closes in a row having heard nothing
|
||||
for ctx.Err() == nil {
|
||||
w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
|
||||
if err != nil {
|
||||
@@ -215,8 +215,10 @@ func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module st
|
||||
return
|
||||
}
|
||||
h.Nudge(module) // a step may have ended while the watch was closing
|
||||
if !heard {
|
||||
h.logf("the watch of the steps of %s closed having heard nothing, so it is left: an ask whose own watch goes deaf is read only at a reconnect", module)
|
||||
if heard {
|
||||
silent = 0
|
||||
} else if silent++; silent >= reopenTries {
|
||||
h.logf("the watch of the steps of %s closed %d times in a row having heard nothing, so it is left: an ask whose own watch goes deaf is read only at a reconnect", module, silent)
|
||||
return
|
||||
}
|
||||
h.logf("the watch of the steps of %s closed; opened again", module)
|
||||
|
||||
Reference in New Issue
Block a user