novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Apply the revision of a composition key whose digest the want names, from the revisions the bucket keeps (issue 1945)

Browse Source
This commit is contained in:
jschoubben
2026-10-25 12:42:20 +01:00
parent 3e3e43de52
commit 8343c4fb36
4 changed files with 163 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-host/nodeagent.go
+14 -3
View File
@@ -294,9 +294,20 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
Haves: sessionHaves{session},
Kept: keptByModule(opts, signer),
Recorded: recordedIn(opts.state, say),
Trusted: trustedKeys(opts, signer),
Backup: backupNode(session),
Say: say,
Revisions: func(ctx context.Context, key string) ([]agent.Update, error) {
got, err := session.KVHistory(ctx, composedBucket, key)
if err != nil {
return nil, err
}
out := make([]agent.Update, 0, len(got))
for _, u := range got {
out = append(out, agent.Update{Key: u.Key, Value: u.Value, Revision: u.Revision, Deleted: u.Deleted})
}
return out, nil
},
Trusted: trustedKeys(opts, signer),
Backup: backupNode(session),
Say: say,
Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result {
actx, done := applyContext(session.WithLink(ctx))
res := applyComposition(actx, opts, c, others, sched, say)
internal/agent/agent.go
+65 -2
View File
@@ -97,7 +97,10 @@ type Env struct {
// Recorded says the engine holds a record of every resource of the composition, so it was applied and not removed
// since; nil says yes. A composition the kept declaration or a have says runs is applied again when it says no.
Recorded func(ctx context.Context, c delivery.Composed) bool
Say func(string)
// Revisions are a composition key's revisions the bucket keeps, oldest first; nil reads none. The composition a want
// names is taken from them when a newer one replaced it under the same key (ADR 0539 clause 1).
Revisions func(ctx context.Context, key string) ([]Update, error)
Say func(string)
}
// Agent follows the wants.
@@ -108,6 +111,9 @@ type Agent struct {
wants map[string]wantNow // by module
comps map[string]compNow // by composed key
running map[string]delivery.Have
// applied is the composition each module was last applied from: a revision a want named from the key's history is
// not the key's latest value, and what it holds is still held out of the others' applies.
applied map[string]compNow
// attempt is the last apply of a module that failed or is held, with the revisions it was made from.
attempt map[string]attempt
checked map[string]bool
@@ -144,7 +150,7 @@ func New(env Env) *Agent {
if env.Say == nil {
env.Say = func(string) {}
}
return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{},
return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{},
attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{},
retry: make(chan []string, 8)}
}
@@ -235,6 +241,8 @@ func (a *Agent) Running() []delivery.Composed {
}
if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed {
out = append(out, c.C)
} else if c, ok := a.applied[m]; ok && c.Dig == h.Composed {
out = append(out, c.C)
}
}
return out
@@ -355,6 +363,51 @@ func (a *Agent) compositionFor(module, fp string) (compNow, bool) {
return c, ok
}
// revisionNamed is the revision of a module's composition key whose digest is the one given, signed by a trusted key,
// from the revisions the bucket keeps; false when none is.
func (a *Agent) revisionNamed(ctx context.Context, module, fp, digest string) (compNow, bool) {
if a.env.Revisions == nil {
return compNow{}, false
}
var key string
var err error
if module == NodeModule {
key, err = delivery.ComposedNodeKey(a.env.Node)
} else {
key, err = delivery.ComposedKey(a.env.Node, module, fp)
}
if err != nil {
return compNow{}, false
}
revs, err := a.env.Revisions(ctx, key)
if err != nil {
a.mu.Lock()
a.sayOnce("history:"+key, fmt.Sprintf("the revisions of %s cannot be read: %v", key, err))
a.mu.Unlock()
return compNow{}, false
}
var trusted []ed25519.PublicKey
if a.env.Trusted != nil {
trusted = a.env.Trusted()
}
for i := len(revs) - 1; i >= 0; i-- {
u := revs[i]
if u.Deleted || u.Key != key {
continue
}
c, err := delivery.DecodeComposed(u.Value)
if err != nil {
continue
}
dig, err := delivery.VerifyComposed(c, trusted)
if err != nil || dig != digest {
continue
}
return compNow{Raw: u.Value, C: c, Dig: dig, Rev: u.Revision}, true
}
return compNow{}, false
}
// others are what the node's other modules run now, so what they hold is not taken: the compositions the agent runs and,
// for every module the agent does not hold yet, the resources the kept declaration has for it (ADR 0429 §4).
func (a *Agent) others(except string) []delivery.Composed {
@@ -373,6 +426,8 @@ func (a *Agent) others(except string) []delivery.Composed {
}
if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed {
out = append(out, c.C)
} else if c, ok := a.applied[m]; ok && c.Dig == h.Composed {
out = append(out, c.C)
}
}
if a.env.Kept != nil {
@@ -511,6 +566,13 @@ func (a *Agent) settleOne(ctx context.Context, module string) {
a.mu.Lock()
comp, have := a.compositionFor(module, w.Want.FP)
a.mu.Unlock()
if (!have || comp.Err != nil || comp.Dig != w.Want.Composed) && w.Want.Composed != "" {
// The composition the want names may be an older revision of the key, replaced by a newer one at the same
// fingerprint (a want held there keeps its digest): taken from the revisions the bucket keeps (ADR 0539 clause 1).
if old, found := a.revisionNamed(ctx, module, w.Want.FP, w.Want.Composed); found {
comp, have = old, true
}
}
if !have || comp.Err != nil {
return // waits for the composition; a change of its key settles the module again
}
@@ -666,6 +728,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
}
delete(a.attempt, module)
a.running[module] = delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning}
a.applied[module] = comp
a.mu.Unlock()
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail})
}
internal/agent/agent_test.go
+26
View File
@@ -809,3 +809,29 @@ func TestACompositionWhoseResourcesWereRemovedIsAppliedAgain(t *testing.T) {
})
}
}
// The composition a want names is applied from the key's history when a newer one has replaced it under the same key:
// a want held at a fingerprint keeps naming the digest it was written with, and the controller writes every new
// composition at that fingerprint over the same key (ADR 0539 clause 1, the sweep's kept revisions; issue 1945, live on
// 2026-10-25: shanks' node-tools composition had digest 575a5c96, its held want named a36e63c8, and nothing was applied).
func TestTheRevisionAWantNamesIsAppliedFromHistory(t *testing.T) {
b := newBed(t)
fp := fpOf("a")
ck, oldRaw, oldDig := composition(t, "node-tools", fp, false, file("node-tools", "x", "old"))
_, newRaw, _ := composition(t, "node-tools", fp, false, file("node-tools", "x", "new"))
b.comps.before(ck, newRaw) // the key's latest value
wk, wraw := wantFor(t, "node-tools", fp, oldDig, "")
b.wants.before(wk, wraw)
b.env.Revisions = func(_ context.Context, key string) ([]Update, error) {
if key != ck {
return nil, nil
}
return []Update{{Key: ck, Value: oldRaw, Revision: 3}, {Key: ck, Value: newRaw, Revision: 7}}, nil
}
stop := b.run()
defer stop()
eventually(t, "the revision the want names was not applied", func() bool { return len(b.appliedList()) == 1 })
if h := b.haves.last("node-tools"); h.Composed != oldDig {
t.Fatalf("the have names %s, want %s", h.Composed, oldDig)
}
}
internal/link/watch.go
+58
View File
@@ -83,6 +83,64 @@ var (
_ KVBus = (*natsLink)(nil)
)
// KVHistoryBus is a link that can read the revisions a bucket keeps of one key: the node agent takes the composition its
// want names from them when a newer one replaced it (novox/hq ADR 0539 clause 1).
type KVHistoryBus interface {
// KVHistory is the key's revisions, oldest first; none when the key has none.
KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error)
}
// KVHistory reads the key's revisions on the link open now.
func (b OverNATS) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
js := b.JS
if b.Conn != nil {
bound, err := b.Conn.JetStream()
if err != nil {
return nil, err
}
js = bound
}
kv, err := js.KeyValue(bucket)
if err != nil {
return nil, fmt.Errorf("the bucket %s: %w", bucket, err)
}
entries, err := kv.History(key, nats.Context(ctx))
if errors.Is(err, nats.ErrKeyNotFound) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, err)
}
out := make([]KVUpdate, 0, len(entries))
for _, e := range entries {
out = append(out, KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge})
}
return out, nil
}
func (l *natsLink) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
return OverNATS{Conn: l.conn, JS: l.js}.KVHistory(ctx, bucket, key)
}
var (
_ KVHistoryBus = OverNATS{}
_ KVHistoryBus = (*natsLink)(nil)
)
// KVHistory reads a key's revisions on the link open now.
func (q *Session) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
q.init()
q.mu.Lock()
bus := q.bus
q.mu.Unlock()
h, ok := bus.(KVHistoryBus)
if bus == nil || !ok {
return nil, ErrNoLink
}
return h.KVHistory(ctx, bucket, key)
}
// ErrNoLink is what a watch asks for with no link open.
var ErrNoLink = errors.New("no link to the bus is open")