Bound a state watch's hand-over by its progress, not its whole set, so a large state's newest keys are handed over (issue 1726)
mesh/delivery delivering: 0 machine step(s) passed
mesh/check pass 0211844e6ad3 on 0e985f9bf186 | unit 0 run 117 kept | int 0 run 119 kept | clone 5.4s | composition passed on 4 machines
mesh/delivery delivering: 0 machine step(s) passed
mesh/check pass 0211844e6ad3 on 0e985f9bf186 | unit 0 run 117 kept | int 0 run 119 kept | clone 5.4s | composition passed on 4 machines
The gitea module's watch of mesh-delivery-checker.owed failed at every attempt since 03:05 on novox: about 200 owed keys, each a round trip to the bundle, outlast a 10 s bound for the whole set. The bundle dropped its handler, the oldest keys were written again every 40 s and the newest, handed over last, never were: group fix/1937-done-key-blobs's pass never reached mesh/check. StateWatch now fails only when no current value is handed over within StateTimeout.
This commit is contained in:
@@ -261,7 +261,12 @@ func (c *Conn) writable(module, name string) (StateIssued, error) {
|
||||
|
||||
// StateWatch hands deliver the current value of every key matching the pattern — none that is
|
||||
// deleted — and then every change, in order (ADR 0201). It returns once the current values are
|
||||
// delivered; deliver is called from one goroutine, one change at a time, and an error from it is said
|
||||
// delivered. **StateTimeout bounds the time without progress, not the whole set** (novox/hq issue
|
||||
// 1726): each current value is a round trip to the bundle, so a state of a few hundred keys outlasts
|
||||
// any fixed bound for all of them; the watch then failed at every attempt, the bundle dropped its
|
||||
// handler, and the newest keys, handed over last, were never taken (the checker's owed, whose newest
|
||||
// verdicts never reached mesh/check). It fails only when no current value arrives, or one is not
|
||||
// taken, within StateTimeout; deliver is called from one goroutine, one change at a time, and an error from it is said
|
||||
// and the watch goes on: state is not a queue, and the next change, or the next start, reads it again.
|
||||
// The pattern is a key, with `*` for one name and `**` for the rest; empty is every key.
|
||||
func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange) error) (stop func(), err error) {
|
||||
@@ -306,6 +311,14 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange
|
||||
})
|
||||
}
|
||||
current := make(chan struct{})
|
||||
// handed says a current value was handed over or passed by: progress, which restarts the bound.
|
||||
handed := make(chan struct{}, 1)
|
||||
progress := func() {
|
||||
select {
|
||||
case handed <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
go func() {
|
||||
initial := true
|
||||
for e := range w.Updates() {
|
||||
@@ -326,6 +339,7 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange
|
||||
// A deletion among the current values is a key that is not there: not handed over
|
||||
// (measured, research 024 — the server sends its marker among the initial values).
|
||||
if initial {
|
||||
progress()
|
||||
continue
|
||||
}
|
||||
change.Op = "delete"
|
||||
@@ -334,17 +348,32 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange
|
||||
c.Logf("[mesh-tools] %s did not take %s %s of %s: %v; its next change, or its next start, reads it again",
|
||||
module, change.Op, change.Key, name, err)
|
||||
}
|
||||
if initial {
|
||||
progress()
|
||||
}
|
||||
}
|
||||
if initial {
|
||||
close(current)
|
||||
}
|
||||
}()
|
||||
select {
|
||||
case <-current:
|
||||
return stop, nil
|
||||
case <-time.After(StateTimeout):
|
||||
stop()
|
||||
return nil, fmt.Errorf("the current values of %s did not arrive in %s", name, StateTimeout)
|
||||
bound := time.NewTimer(StateTimeout)
|
||||
defer bound.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-current:
|
||||
return stop, nil
|
||||
case <-handed:
|
||||
if !bound.Stop() {
|
||||
select {
|
||||
case <-bound.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
bound.Reset(StateTimeout)
|
||||
case <-bound.C:
|
||||
stop()
|
||||
return nil, fmt.Errorf("the current values of %s did not arrive in %s: none was handed over in that time", name, StateTimeout)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user