diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index a595f23c..93d3a943 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -258,6 +258,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // reads it and drops a retired node by hand. reads.ctx = context.WithoutCancel(ctx) reads.window = &oldForm{forms: forms.Current, findings: found, logf: logger.Printf, now: time.Now, log: func(e delivery.BuildLogEntry) { b.Log(e) }, + bounds: pubBoundsKV.KV, said: func(body versions.OldFormClosedBody) { if err := emitEvent(context.WithoutCancel(ctx), js, cred, versions.OldFormClosedEvent, "old-form-closed-"+body.At.Format(time.RFC3339Nano), body); err != nil { logger.Printf("saying %s: %v", versions.OldFormClosedEvent, err) @@ -371,7 +372,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } }) pubBoundsKV.Ready(func(kv jetstream.KeyValue) { - each := func(e jetstream.KeyValueEntry) { pubs.ended(ctx, e) } + each := func(e jetstream.KeyValueEntry) { + if versions.IsOldFormBound(e.Key()) { + if e.Operation() == jetstream.KeyValuePurge { + reads.stepWindow() // a node's seven days ended: the step raises old-form-held now + } + return + } + pubs.ended(ctx, e) + } if pw, err := settle(ctx, kv, each); err != nil { process.NotWatched(logger.Printf, versions.BoundsBucket, err, "a publish no holder answers, and a node holding the old form seven days, are not raised until the builder starts again") } else { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/oldform.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/oldform.go index edabe022..24d6c32a 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/oldform.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/oldform.go @@ -34,6 +34,10 @@ type oldForm struct { logf func(string, ...any) now func() time.Time + // bounds is the builder's state bounds: each node holding the window has a key there whose time to live ends at its + // seven days, so old-form-held is raised then, not at the next change (review round 1); nil keeps none. + bounds func() jetstream.KeyValue + mu sync.Mutex w versions.OldFormWindow kv jetstream.KeyValue @@ -106,6 +110,7 @@ func (o *oldForm) step(ctx context.Context, named []string, counted map[string][ needs: "builder_old_form shows the window; a retired node is dropped there by hand; builder_engine_reads asks the node again."}) } o.overdue = overdue + o.keepBounds(ctx, before, next, overdue) if closed { body := next.Closing() line := fmt.Sprintf("the old reference form's window closed at %s: no counted node-engine reads only an older form (last held by %s; dropped by hand: %d); new build keys leave it out", @@ -123,6 +128,37 @@ func (o *oldForm) step(ctx context.Context, named []string, counted map[string][ } } +// keepBounds writes a bound for each node holding the window and not yet overdue, ending at its seven days, and deletes +// the bound of each node that no longer holds it. A standing bound keeps its time (a Create). Called with mu held. +func (o *oldForm) keepBounds(ctx context.Context, before, next versions.OldFormWindow, overdue []string) { + if o.bounds == nil { + return + } + kv := o.bounds() + if kv == nil { + return + } + for _, n := range before.Holding { + if !slices.Contains(next.Holding, n) { + if err := kv.Delete(ctx, versions.OldFormBoundKey(n)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + o.logf("the old form's bound of %s is not deleted: %v", n, err) + } + } + } + for _, n := range next.Holding { + left := next.Since[n].Add(versions.FindingAfter).Sub(o.now()) + if slices.Contains(overdue, n) || left <= 0 { + continue + } + if left < time.Second { + left = time.Second // the bus's least time to live + } + if _, err := kv.Create(ctx, versions.OldFormBoundKey(n), []byte(`{}`), jetstream.KeyTTL(left)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) { + o.logf("the old form's bound of %s is not written, so its seven days are seen at the next change only: %v", n, err) + } + } +} + // drop is builder_old_form's drop: the node leaves the count, the window is stepped by the caller. func (o *oldForm) drop(ctx context.Context, node, by, why string, at time.Time) (string, error) { o.mu.Lock() diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go index f038ea5f..1494f272 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -393,7 +393,8 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { fs := forms.ArtifactForms{{Name: "registry", First: true, Kinds: []string{"image"}}, {Name: "oci", Kinds: []string{"image"}}} var closings []versions.OldFormClosedBody var logged []delivery.BuildLogEntry - window := &oldForm{forms: fs, findings: f, logf: t.Logf, now: func() time.Time { return now }, + marks := &memBounds{keys: map[string]bool{}} + window := &oldForm{forms: fs, findings: f, logf: t.Logf, now: func() time.Time { return now }, bounds: func() jetstream.KeyValue { return marks }, said: func(b versions.OldFormClosedBody) { closings = append(closings, b) }, log: func(e delivery.BuildLogEntry) { logged = append(logged, e) }} reads := &engineReads{keeper: &enginereads.Keeper{}, window: window, counted: []string{"a", "b"}} @@ -404,6 +405,11 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { if !window.w.Open || !slices.Equal(window.w.Holding, []string{"b"}) || st.vals[heldKey] != nil { t.Fatalf("one node lacking the newest form: %+v", window.w) } + // Review round 1: the node holding it has a bound in the state bounds ending at its seven days, so old-form-held is + // raised then, not at the next change. + if !marks.keys[versions.OldFormBoundKey("b")] || marks.keys[versions.OldFormBoundKey("a")] { + t.Fatalf("the old form's bounds = %v, want b's alone", marks.keys) + } // Security-upgrade F3 (review round 1): an empty node_haves read, naming no node, leaves the window open. window.step(context.Background(), nil, map[string][]string{}) if !window.w.Open || !slices.Equal(window.w.Holding, []string{"b"}) || len(closings) != 0 { @@ -417,6 +423,9 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { counted["b"] = newest window.step(context.Background(), []string{"a", "b"}, counted) window.step(context.Background(), []string{"a", "b"}, counted) + if marks.keys[versions.OldFormBoundKey("b")] { + t.Error("a node no longer holding the window keeps its bound") + } if window.w.Open || len(closings) != 1 || !slices.Equal(closings[0].LastHolding, []string{"b"}) || st.vals[heldKey] != nil { t.Errorf("the node back with the newest form: %+v, closings %+v, finding left %v", window.w, closings, st.vals[heldKey] != nil) } @@ -453,6 +462,24 @@ func (m memVersions) KeysOf(context.Context, string) ([]string, error) { return slices.Collect(maps.Keys(m)), nil } +// memBounds is the state bounds in memory: which keys stand. +type memBounds struct { + jetstream.KeyValue + keys map[string]bool +} + +func (m *memBounds) Create(_ context.Context, key string, _ []byte, _ ...jetstream.KVCreateOpt) (uint64, error) { + if m.keys[key] { + return 0, jetstream.ErrKeyExists + } + m.keys[key] = true + return 1, nil +} +func (m *memBounds) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { + delete(m.keys, key) + return nil +} + // purged is the bus's marker of a key whose time to live ended. type purged struct { jetstream.KeyValueEntry