diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index fe2dd79..ba3a246 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -431,6 +431,19 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { k, _ := delivery.PlacementCandidateKey("same", "n1") sb.del(delivery.PlacementCandidatesBucket, k) waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 3 }) + + // Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches + // again. The new run reads afresh, so that group is no longer answered. + sb.missing(delivery.PlacementCandidatesBucket, true) + sb.end(delivery.ModulesBucket) + dropsKey, _ := delivery.PlacementCandidateKey("drops", "n1") + sb.del(delivery.PlacementCandidatesBucket, dropsKey) + time.Sleep(30 * time.Millisecond) + sb.missing(delivery.PlacementCandidatesBucket, false) + waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool { + _, ok := a["drops"] + return !ok && len(a) == 2 + }) } // patternBus is a bus whose every watch gets the updates of its bucket matching its pattern, as the bus's own watches @@ -439,6 +452,26 @@ type patternBus struct { mu sync.Mutex watches []patternWatch absent map[string]bool + // kv is each bucket's keys now; once replay is set, a new watch gets those matching it and then Caught, as a watch + // of the bus's own does. + kv map[string]map[string]agent.Update + replay bool +} + +// end ends every watch of a bucket, as a watch the bus drops, and from then on each new watch replays what is there. +func (b *patternBus) end(bucket string) { + b.mu.Lock() + defer b.mu.Unlock() + b.replay = true + kept := b.watches[:0] + for _, w := range b.watches { + if w.bucket == bucket { + close(w.ch) + continue + } + kept = append(kept, w) + } + b.watches = kept } // missing makes a bucket's watch fail, as one the bus holds no state for, or work again. @@ -466,6 +499,14 @@ func (b *patternBus) watch(_ context.Context, bucket, pattern string) (<-chan ag } w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)} b.watches = append(b.watches, w) + if b.replay { + for key, u := range b.kv[bucket] { + if matches(pattern, key) { + w.ch <- u + } + } + w.ch <- agent.Update{Caught: true} + } return w.ch, nil } @@ -485,9 +526,25 @@ func matches(pattern, key string) bool { func (b *patternBus) send(bucket string, u agent.Update) { b.mu.Lock() defer b.mu.Unlock() + if !u.Caught { + if b.kv == nil { + b.kv = map[string]map[string]agent.Update{} + } + if b.kv[bucket] == nil { + b.kv[bucket] = map[string]agent.Update{} + } + if u.Deleted { + delete(b.kv[bucket], u.Key) + } else { + b.kv[bucket][u.Key] = u + } + } for _, w := range b.watches { if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) { - w.ch <- u + select { + case w.ch <- u: + default: // a watch of a run that ended reads no more + } } } } diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index cef381c..2bbdb1d 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -78,6 +78,14 @@ func (s *Shadow) Keep(ctx context.Context, backoff func(try int) time.Duration) // Run answers until ctx ends or an input's watch ends. func (s *Shadow) Run(ctx context.Context) error { + // Each run watches afresh and starts from nothing read: what changed while it was down (a candidate key deleted) + // is then not missed, and the watches of a run before are not left open (review of L2b, round 2). + ctx, cancel := context.WithCancel(ctx) + defer cancel() + s.mu.Lock() + s.candidates = map[string]map[string]candidateKey{} + s.c = New(s.c.env) + s.mu.Unlock() c := s.c node := c.env.Node ins := []input{