F-delivery 6: old-form-held is raised at a node's seventh day by an expiring key
The finding was written only at the window's next change. Each node holding the window has a key of the builder's state bounds whose time to live ends at its seven days; the bus's marker of its end steps the window, which raises the finding then; a node that no longer holds it loses its key. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user