novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Assigner: its asks and their keys are its asks package's, an asking's id the delivery client's, the view's answers the view's (ADR 0634 §5; issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 15:54:19 +01:00
parent 54f91d5ad9
commit 44397a53fc
12 changed files with 123 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go
+12 -10
View File
@@ -9,7 +9,9 @@ import (
"fmt"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"io"
@@ -699,9 +701,9 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) {
!strings.HasPrefix(h.Words, "health failed, phase running, seen at ") || !strings.HasSuffix(h.Words, ": connection refused") {
t.Fatalf("held = %+v, %v", h, err)
}
akey, _ := delivery.HoldAskKey("web", fNew)
akey, _ := asks.HoldAskKey("web", fNew)
a, err := decodeAsk(akey, b.get(t, buckets.Asks, akey))
if err != nil || a.Asking <= 0 || a.Ask.ID != delivery.AskingID(akey, a.Asking) || len(a.Ask.Options) != 2 {
if err != nil || a.Asking <= 0 || a.Ask.ID != meshdeliveryclient.AskingID(akey, a.Asking) || len(a.Ask.Options) != 2 {
t.Fatalf("the ask = %+v, %v", a, err)
}
}
@@ -755,7 +757,7 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) {
b.place(t, "ace", "mail")
b.start(t)
akey, _ := delivery.RestoreAskKey("ace", "mail", fNew)
akey, _ := asks.RestoreAskKey("ace", "mail", fNew)
a, err := decodeAsk(akey, b.get(t, buckets.Asks, akey))
if err != nil || len(a.Ask.Options) != 2 || a.Ask.Options[0].ID != "restore" || a.Ask.Options[1].ID != "leave-stopped" {
t.Fatalf("the restore ask = %+v, %v", a, err)
@@ -924,7 +926,7 @@ func TestAReleaseAnswerWritesTheWantAgain(t *testing.T) {
if w.Newest != fNew {
t.Fatalf("the set-back want = %+v", w)
}
hold, _ := delivery.HoldAskKey("web", fNew)
hold, _ := asks.HoldAskKey("web", fNew)
b.get(t, buckets.Asks, hold)
b.answer(t, hold, "release")
w = b.wantNow(t, "ace", "web", "the held fingerprint's want written again", func(w delivery.Want) bool { return w.FP == fNew })
@@ -969,7 +971,7 @@ func TestABackupAnswerIsWrittenIntoTheNodesWant(t *testing.T) {
held, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fNew, Phase: nodeengineclient.PhaseHeld, Health: nodeengineclient.HealthWaits, Detail: "no space left"})
b.putJSON(t, buckets.Have, "ace.web", held)
b.start(t)
akey, _ := delivery.BackupAskKey("ace", "web", fNew)
akey, _ := asks.BackupAskKey("ace", "web", fNew)
a := b.askNow(t, akey, 0)
if !strings.Contains(a.Ask.Explanation, "no space left") {
t.Fatalf("the backup ask = %+v", a)
@@ -1014,7 +1016,7 @@ func TestAHoldIsLiftedOnceEveryNodeRunsItHealthy(t *testing.T) {
b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)})
b.start(t)
hkey, _ := delivery.HeldKey("web", fNew)
akey, _ := delivery.HoldAskKey("web", fNew)
akey, _ := asks.HoldAskKey("web", fNew)
b.get(t, buckets.Wants, hkey)
b.get(t, buckets.Asks, akey)
healthy, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fNew, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
@@ -1048,7 +1050,7 @@ func TestAHoldWrittenAgainAfterALiftAndARestartIsAskedAgain(t *testing.T) {
b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)})
_, stop := b.startStoppable(t)
hkey, _ := delivery.HeldKey("web", fNew)
akey, _ := delivery.HoldAskKey("web", fNew)
akey, _ := asks.HoldAskKey("web", fNew)
first, err := decodeAsk(akey, b.get(t, buckets.Asks, akey))
if err != nil {
t.Fatal(err)
@@ -1242,7 +1244,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
}
answerAll(p)
p = proposalNow(p.ID, "asked", func(p mma.Proposal) bool { return p.State == mmac.ProposalAsked })
akey, _ := delivery.PlacementAskKey(p.ID)
akey, _ := asks.PlacementAskKey(p.ID)
b.answer(t, akey, assign.OptionYes)
select {
case e := <-heard:
@@ -1286,7 +1288,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
})
answerAll(relaid)
proposalNow(moved.ID, "asked", func(p mma.Proposal) bool { return p.State == mmac.ProposalAsked })
akey, _ = delivery.PlacementAskKey(moved.ID)
akey, _ = asks.PlacementAskKey(moved.ID)
b.answer(t, akey, assign.OptionNo)
for deadline := time.Now().Add(15 * time.Second); b.hasKey(plc.ProposalsBucket, moved.ID); time.Sleep(50 * time.Millisecond) {
if time.Now().After(deadline) {
@@ -1338,7 +1340,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
if ace, err := b.buckets[plc.PlacementBucket].Get(b.ctx, "ace.node"); err != nil || !strings.Contains(string(ace.Value()), "10.9.0.2") {
t.Errorf("the joining node's key = %v, %v", ace, err)
}
if akey, _ := delivery.PlacementAskKey(joined.(proposed).Proposal.ID); b.hasKey(buckets.Asks, akey) || b.hasKey(plc.ProposalsBucket, joined.(proposed).Proposal.ID) {
if akey, _ := asks.PlacementAskKey(joined.(proposed).Proposal.ID); b.hasKey(buckets.Asks, akey) || b.hasKey(plc.ProposalsBucket, joined.(proposed).Proposal.ID) {
t.Error("an ask was raised for the joining node, or its proposal stays open")
}
// Its enrolment ends before it confirms: a remove-node on the same warrant deletes its key at once, with no ask, and
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go
+15 -14
View File
@@ -7,6 +7,7 @@ import (
"errors"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"maps"
"mesh-delivery-assigner/internal/buckets"
@@ -214,18 +215,18 @@ func (p *passer) wants(ctx context.Context) {
return err == nil && p.w.has(grants, key)
},
Restored: func(node string) bool {
key, err := delivery.RestoreAskKey(node, name, fp)
key, err := asks.RestoreAskKey(node, name, fp)
return err == nil && p.answeredWith(key, assign.OptionRestore)
},
Allowed: func(node, digest string) bool {
key, err := delivery.RecordAskKey(node, name, digest)
key, err := asks.RecordAskKey(node, name, digest)
return err == nil && p.answeredWith(key, assign.OptionYes)
},
Release: func(fp string) (string, bool) { return p.w.release(name, fp) }}
if fp != "" {
held, _ := delivery.HeldKey(name, fp)
grant, _ := grantsstate.PartGrantKey(name, fp)
hold, _ := delivery.HoldAskKey(name, fp)
hold, _ := asks.HoldAskKey(name, fp)
m.Released = p.answeredWith(hold, "release")
m.Kept = p.answeredWith(hold, "keep")
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(buckets.Wants, held) && !m.Released, p.w.has(grants, grant)
@@ -549,7 +550,7 @@ func (p *passer) put(ctx context.Context, bucket, key string, encode func() ([]b
// no proof of who answered, which the view may answer and the teller fills. In release 5 an answer changes nothing on a
// node (wants are only compared, ADR 0429 §7), and the options say so.
func (p *passer) holdAsk(ctx context.Context, module, fp string, fresh *delivery.Held) {
key, err := delivery.HoldAskKey(module, fp)
key, err := asks.HoldAskKey(module, fp)
if err != nil || p.w.has(buckets.Asks, key) {
return
}
@@ -570,7 +571,7 @@ func (p *passer) lift(ctx context.Context, module, fp string) {
if err != nil {
return
}
ask, err := delivery.HoldAskKey(module, fp)
ask, err := asks.HoldAskKey(module, fp)
if err != nil {
return
}
@@ -592,7 +593,7 @@ func (p *passer) lift(ctx context.Context, module, fp string) {
// The restore asks of that fingerprint, on every node, go with the hold: every node runs it healthy, so no node waits to be
// set back from it (issue 1906, note 28).
for key := range p.w.read(buckets.Asks) {
if k, err := delivery.ParseAnyAskKey(key); err == nil && k.Kind == delivery.AskKindRestore && k.Module == module && k.FP == fp {
if k, err := asks.ParseAnyAskKey(key); err == nil && k.Kind == asks.AskKindRestore && k.Module == module && k.FP == fp {
p.remove(ctx, buckets.Asks, key)
}
}
@@ -604,15 +605,15 @@ func (p *passer) lift(ctx context.Context, module, fp string) {
// A module of the trunk whose fingerprint is not read yet, and a composition not read yet, are not stale: nothing known is
// nothing taken away.
func (p *passer) staleAsk(key string, trunk map[string]meshcatalog.ModulesByModuleValue) bool {
k, err := delivery.ParseAnyAskKey(key)
k, err := asks.ParseAnyAskKey(key)
if err != nil {
return false
}
switch k.Kind {
case delivery.AskKindHold, delivery.AskKindRestore:
case asks.AskKindHold, asks.AskKindRestore:
m, on := trunk[k.Module]
return !on || (m.Fp2 != "" && m.Fp2 != k.FP)
case delivery.AskKindRecord:
case asks.AskKindRecord:
m, on := trunk[k.Module]
if !on {
return true
@@ -674,8 +675,8 @@ func (p *passer) staleAsks(ctx context.Context, trunk map[string]meshcatalog.Mod
if p.staleAsk(key, trunk) {
// The condition is cleared first and the ask goes only once that was said: the ask is what brings this key back at the
// next pass, so a clear that failed once is made again, with no state of its own (issue 1906, note 29).
k, kerr := delivery.ParseAnyAskKey(key)
held := kerr == nil && (k.Kind == delivery.AskKindHold || k.Kind == delivery.AskKindRestore)
k, kerr := asks.ParseAnyAskKey(key)
held := kerr == nil && (k.Kind == asks.AskKindHold || k.Kind == asks.AskKindRestore)
if held && !p.clearHeld(k.Module, k.FP) {
continue
}
@@ -736,7 +737,7 @@ func (p *passer) heldCondition(ctx context.Context, module, fp string, fresh *de
// restoreAsk writes, once, the ask about a module holding data whose held fingerprint failed its health check on node.
func (p *passer) restoreAsk(ctx context.Context, module, node, fp string) {
key, err := delivery.RestoreAskKey(node, module, fp)
key, err := asks.RestoreAskKey(node, module, fp)
if err != nil || p.w.has(buckets.Asks, key) {
return
}
@@ -779,7 +780,7 @@ func (p *passer) clearHeld(module, fp string) bool {
// recordAsk writes, once, the ask about a move of a module of policy record that waits for the operator's yes.
func (p *passer) recordAsk(ctx context.Context, module, fp string, need assign.RecordNeed) {
key, err := delivery.RecordAskKey(need.Node, module, need.Digest)
key, err := asks.RecordAskKey(need.Node, module, need.Digest)
if err != nil || p.w.has(buckets.Asks, key) {
return
}
@@ -811,7 +812,7 @@ func (p *passer) backups(ctx context.Context) {
if !ok || want.Removed {
continue
}
akey, err := delivery.BackupAskKey(node, module, want.FP)
akey, err := asks.BackupAskKey(node, module, want.FP)
if err != nil {
continue
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+24 -23
View File
@@ -10,6 +10,7 @@ import (
watchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"mesh-delivery-assigner/internal/buckets"
@@ -101,7 +102,7 @@ var fpTest = meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64)
func TestAHoldAskedAgainAfterALiftAndARestartHasANewID(t *testing.T) {
asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}}
kvs := kvSet{buckets.Asks: asksKV, buckets.Wants: wantsKV}
key, _ := delivery.HoldAskKey("web", fpTest)
key, _ := asks.HoldAskKey("web", fpTest)
held := &delivery.Held{Node: "ace", Words: "health failed"}
first := newPasser(kvs, time.Date(2026, 10, 25, 1, 0, 0, 0, time.UTC))
first.holdAsk(context.Background(), "web", fpTest, held)
@@ -135,7 +136,7 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) {
t.Errorf("offered(%s) = %v, want %v", provision, !want, want)
}
}
key, _ := delivery.HoldAskKey("web", fpTest)
key, _ := asks.HoldAskKey("web", fpTest)
p := newPasser(kvSet{}, time.Now())
put := func(asking int) string {
a := assign.HoldAsk(key, "web", fpTest, "ace", "health failed", asking, time.Now())
@@ -404,7 +405,7 @@ func putOfTheBus(t *testing.T) {
g, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
world(buckets.Grants, gkey, g)
got := step()
akey, _ := delivery.RecordAskKey("novox", "nats", digest)
akey, _ := asks.RecordAskKey("novox", "nats", digest)
if got.Step != delivery.PutPerson || got.Ask != akey {
t.Fatalf("composed, the put stands at %s (%s), ask %q", got.Step, got.Words, got.Ask)
}
@@ -751,14 +752,14 @@ func TestAsksOfAFingerprintTheTrunkLeftBehindAreTakenAway(t *testing.T) {
return map[string]meshcatalog.ModulesByModuleValue{"web": {Fp2: fp, Manifest: "sha256:" + strings.Repeat("1", 64), Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}}
}
type ask struct{ name, key string }
holdOld, _ := delivery.HoldAskKey("web", fpOld)
holdNew, _ := delivery.HoldAskKey("web", fpNew)
holdGone, _ := delivery.HoldAskKey("gone", fpOld)
restOld, _ := delivery.RestoreAskKey("novox", "web", fpOld)
restNew, _ := delivery.RestoreAskKey("novox", "web", fpNew)
recOld, _ := delivery.RecordAskKey("novox", "web", digOld)
recNew, _ := delivery.RecordAskKey("novox", "web", digNew)
backOld, _ := delivery.BackupAskKey("novox", "web", fpOld)
holdOld, _ := asks.HoldAskKey("web", fpOld)
holdNew, _ := asks.HoldAskKey("web", fpNew)
holdGone, _ := asks.HoldAskKey("gone", fpOld)
restOld, _ := asks.RestoreAskKey("novox", "web", fpOld)
restNew, _ := asks.RestoreAskKey("novox", "web", fpNew)
recOld, _ := asks.RecordAskKey("novox", "web", digOld)
recNew, _ := asks.RecordAskKey("novox", "web", digNew)
backOld, _ := asks.BackupAskKey("novox", "web", fpOld)
all := []string{holdOld, holdNew, holdGone, restOld, restNew, recOld, recNew, backOld}
for _, c := range []struct {
name string
@@ -778,14 +779,14 @@ func TestAsksOfAFingerprintTheTrunkLeftBehindAreTakenAway(t *testing.T) {
var said []oc.Condition
p.say = func(c oc.Condition) error { said = append(said, c); return nil }
for _, key := range all {
k, _ := delivery.ParseAnyAskKey(key)
var a delivery.AssignerAsk
k, _ := asks.ParseAnyAskKey(key)
var a asks.Ask
switch k.Kind {
case delivery.AskKindHold:
case asks.AskKindHold:
a = assign.HoldAsk(key, k.Module, k.FP, "novox", "w", 1, time.Now())
case delivery.AskKindRestore:
case asks.AskKindRestore:
a = assign.RestoreAsk(key, k.Node, k.Module, k.FP, "w", time.Now())
case delivery.AskKindRecord:
case asks.AskKindRecord:
a = assign.RecordAsk(key, k.Node, k.Module, fpNew, assign.RecordNeed{Node: k.Node, Digest: k.Composed}, time.Now())
default:
a = assign.BackupAsk(key, k.Node, k.Module, k.FP, digOld, "w", 2, time.Now())
@@ -869,14 +870,14 @@ func TestAsksOfAFingerprintTheTrunkLeftBehindAreTakenAway(t *testing.T) {
t.Run("a lift takes the restore asks of its fingerprint on every node", func(t *testing.T) {
asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}}
p := newPasser(kvSet{buckets.Asks: asksKV, buckets.Wants: wantsKV}, time.Now())
keep, _ := delivery.RestoreAskKey("ace", "web", fpNew)
keepOther, _ := delivery.RestoreAskKey("ace", "other", fpOld)
for _, key := range []string{restOld, mustKey(delivery.RestoreAskKey("ace", "web", fpOld)), keep, keepOther} {
keep, _ := asks.RestoreAskKey("ace", "web", fpNew)
keepOther, _ := asks.RestoreAskKey("ace", "other", fpOld)
for _, key := range []string{restOld, mustKey(asks.RestoreAskKey("ace", "web", fpOld)), keep, keepOther} {
asksKV.values[key] = []byte("x")
p.w.put(buckets.Asks, key, []byte("x"))
}
p.lift(context.Background(), "web", fpOld)
for key, gone := range map[string]bool{restOld: true, mustKey(delivery.RestoreAskKey("ace", "web", fpOld)): true, keep: false, keepOther: false} {
for key, gone := range map[string]bool{restOld: true, mustKey(asks.RestoreAskKey("ace", "web", fpOld)): true, keep: false, keepOther: false} {
if _, there := asksKV.values[key]; there == gone {
t.Errorf("%s: left in the bucket = %v after the lift", key, there)
}
@@ -960,7 +961,7 @@ func aReleasesApproveAnswersTheMoveOfItsBuildAlone(t *testing.T) {
if got := w.Yes == waiting && w.Release == "feat/web"; got != c.yes || (!c.yes && (w.Yes != "" || w.Release != "")) {
t.Errorf("%s: the want is %+v", c.name, w)
}
akey, _ := delivery.RecordAskKey("g14", "web", waiting)
akey, _ := asks.RecordAskKey("g14", "web", waiting)
if _, there := asksKV.values[akey]; there != c.asked {
t.Errorf("%s: a record ask raised %v, want %v", c.name, there, c.asked)
}
@@ -1026,7 +1027,7 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) {
waits, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpOld, Composed: was, Phase: nodeengineclient.PhaseWaitingYes, Health: nodeengineclient.HealthOK,
Judgement: assign.JudgedRecreates, Waits: waiting, Detail: "it recreates web.server"})
put(buckets.Have, "g14.web", waits)
akey, _ := delivery.RecordAskKey("g14", "web", waiting)
akey, _ := asks.RecordAskKey("g14", "web", waiting)
p.wants(context.Background())
if _, there := asksKV.values[akey]; !there {
t.Fatal("a have waiting for a yes raised no record ask")
@@ -1175,7 +1176,7 @@ func decodeAsk(key string, data []byte) (decodedAsk, error) {
return decodedAsk{Asking: a.Asking, Ask: ask}, err
}
func mustAskID(a delivery.AssignerAsk) string {
func mustAskID(a asks.Ask) string {
ask, err := assign.AskOf(a)
if err != nil {
panic(err)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go
+4 -2
View File
@@ -6,6 +6,8 @@ import (
"errors"
"fmt"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"mesh-delivery-assigner/internal/buckets"
@@ -158,7 +160,7 @@ func (p *passer) puts(ctx context.Context) {
}
}
if next.Warranted {
p.acts.warrant(id, delivery.AskingID(next.Put.Ask, assign.PutAsking(next.Put)))
p.acts.warrant(id, meshdeliveryclient.AskingID(next.Put.Ask, assign.PutAsking(next.Put)))
}
}
}
@@ -197,7 +199,7 @@ func (p *passer) putWorld(node, module string, put delivery.Put, manifest catalo
policy, _, _ = catalogue.DerivedUpgrade(manifest)
}
w.Recorded = policy == catalogue.PolicyRecord
if akey, err := delivery.RecordAskKey(node, module, d.Digest); err == nil && put.Ask == akey {
if akey, err := asks.RecordAskKey(node, module, d.Digest); err == nil && put.Ask == akey {
// Only this put's own ask counts: an ask (and its answer) of an earlier put under the same key is another asking.
if a, ask, err := assign.DecodeAsk(akey, p.w.value(buckets.Asks, akey)); err == nil && a.Asking == assign.PutAsking(put) {
switch {
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go
+3 -2
View File
@@ -8,6 +8,7 @@ import (
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"maps"
"mesh-delivery-assigner/internal/buckets"
"reflect"
@@ -282,7 +283,7 @@ func (p *passer) stepProposal(ctx context.Context, prop mma.Proposal, rev uint64
p.addNode(ctx, prop, revs)
return
}
akey, _ := delivery.PlacementAskKey(prop.ID)
akey, _ := asks.PlacementAskKey(prop.ID)
again := prop.Again != nil && !prop.Again.IsZero() && (prop.Verdict == nil || prop.Again.After(prop.Verdict.At))
if prop.State == mmac.ProposalAsked {
askPulse, _ := plc.PulseAskKey(prop.ID)
@@ -627,7 +628,7 @@ func (p *passer) sayChanged(prop mma.Proposal, answer delivery.Answer, changes [
// endProposal ends a proposal: its ask, its candidates and its key deleted, and placement-withdrawn said carrying it
// whole with why it ended.
func (p *passer) endProposal(ctx context.Context, prop mma.Proposal, why string) {
if akey, err := delivery.PlacementAskKey(prop.ID); err == nil && p.w.has(buckets.Asks, akey) {
if akey, err := asks.PlacementAskKey(prop.ID); err == nil && p.w.has(buckets.Asks, akey) {
if err := p.kvs[buckets.Asks].Delete(ctx, akey); err == nil {
p.w.put(buckets.Asks, akey, nil)
}
modules/mesh-delivery-assigner/go.mod
+6
View File
@@ -56,3 +56,9 @@ require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0
require git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate v0.0.0-00010101000000-000000000000
require git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshnodeenroller v0.0.0-00010101000000-000000000000
require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks v0.0.0-00010101000000-000000000000
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks => ./asks
require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate v0.0.0-00010101000000-000000000000
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate => ../mesh-delivery-view/viewstate
modules/mesh-delivery-assigner/internal/assign/askjson.go
+17 -16
View File
@@ -3,24 +3,25 @@ package assign
import (
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"time"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// assignerAsk is the asking-th ask a, as the asks state holds it: the router's ask as JSON (delivery.AssignerAsk.Ask).
func assignerAsk(asking int, a oc.Ask) delivery.AssignerAsk {
// assignerAsk is the asking-th ask a, as the asks state holds it: the router's ask as JSON (asks.Ask.Ask).
func assignerAsk(asking int, a oc.Ask) asks.Ask {
raw, err := json.Marshal(a)
if err != nil { // an oc.Ask holds only strings, bools and a slice of them: it cannot fail
panic(fmt.Sprintf("assign: the ask %q does not marshal: %v", a.ID, err))
}
return delivery.AssignerAsk{Asking: asking, Ask: raw}
return asks.Ask{Asking: asking, Ask: raw}
}
// AskOf is the router's ask an asks-state value carries.
func AskOf(a delivery.AssignerAsk) (oc.Ask, error) {
func AskOf(a asks.Ask) (oc.Ask, error) {
var ask oc.Ask
if err := json.Unmarshal(a.Ask, &ask); err != nil {
return oc.Ask{}, fmt.Errorf("assign: the value carries no router's ask: %w", err)
@@ -28,32 +29,32 @@ func AskOf(a delivery.AssignerAsk) (oc.Ask, error) {
return ask, nil
}
// EncodeAsk is delivery.EncodeAssignerAsk at key, after the checks every keeper of an ask owes (occ.Carried, ADR 0627 §4):
// the ask's id is delivery.AskingID(key, asking), and it offers at least one option.
func EncodeAsk(key string, a delivery.AssignerAsk) ([]byte, error) {
// EncodeAsk is asks.EncodeAsk at key, after the checks every keeper of an ask owes (occ.Carried, ADR 0627 §4):
// the ask's id is meshdeliveryclient.AskingID(key, asking), and it offers at least one option.
func EncodeAsk(key string, a asks.Ask) ([]byte, error) {
if _, err := checkedAsk(key, a); err != nil {
return nil, err
}
return delivery.EncodeAssignerAsk(key, a)
return asks.EncodeAsk(key, a)
}
func checkedAsk(key string, a delivery.AssignerAsk) (oc.Ask, error) {
ask, err := occ.Carried(a.Ask, delivery.AskingID(key, a.Asking))
func checkedAsk(key string, a asks.Ask) (oc.Ask, error) {
ask, err := occ.Carried(a.Ask, meshdeliveryclient.AskingID(key, a.Asking))
if err != nil {
return oc.Ask{}, fmt.Errorf("assign: %w", err)
}
return ask, nil
}
// DecodeAsk reads the ask at key (delivery.DecodeAssignerAsk) with the router's ask it carries.
func DecodeAsk(key string, data []byte) (delivery.AssignerAsk, oc.Ask, error) {
a, err := delivery.DecodeAssignerAsk(key, data)
// DecodeAsk reads the ask at key (asks.DecodeAsk) with the router's ask it carries.
func DecodeAsk(key string, data []byte) (asks.Ask, oc.Ask, error) {
a, err := asks.DecodeAsk(key, data)
if err != nil {
return delivery.AssignerAsk{}, oc.Ask{}, err
return asks.Ask{}, oc.Ask{}, err
}
ask, err := checkedAsk(key, a)
if err != nil {
return delivery.AssignerAsk{}, oc.Ask{}, err
return asks.Ask{}, oc.Ask{}, err
}
return a, ask, nil
}
modules/mesh-delivery-assigner/internal/assign/asks.go
+13 -11
View File
@@ -3,6 +3,8 @@ package assign
import (
"fmt"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"regexp"
"strings"
"time"
@@ -12,13 +14,13 @@ import (
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// HoldAsk is the ask about a module's fingerprint the assigner holds, at its key (delivery.HoldAskKey): keep it held or
// HoldAsk is the ask about a module's fingerprint the assigner holds, at its key (asks.HoldAskKey): keep it held or
// release it. Releasing writes the held fingerprint's want again by the rules (ADR 0539 §19). node and words are the held
// record's: the machine whose health check failed and the hold's words (issue 1906). asking numbers the writing: 0 the first, a
// larger number an ask written again after a hold lifted by itself, so the router takes it as new.
func HoldAsk(key, module, fp, node, words string, asking int, now time.Time) delivery.AssignerAsk {
func HoldAsk(key, module, fp, node, words string, asking int, now time.Time) asks.Ask {
return assignerAsk(asking, oc.Ask{
ID: delivery.AskingID(key, asking),
ID: meshdeliveryclient.AskingID(key, asking),
Headline: fmt.Sprintf("%s is held at %s", module, short(fp)),
Explanation: fmt.Sprintf("The health check of %s failed on %s at %s (%s), so the assigner holds that version and wants "+
"no node to move to it. It lifts the hold by itself once every node that runs %s runs it healthy; releasing it "+
@@ -43,9 +45,9 @@ const (
// RestoreAsk is the ask about a module holding data whose fingerprint failed its health check on a node (ADR 0429 §6): it is
// not set back by itself, since a set-back would lose the writes made since. words is the check's words. Restore sets the
// node back to the last healthy fingerprint and loses the newer writes; leaving it stopped keeps the data as it is.
func RestoreAsk(key, node, module, fp, words string, now time.Time) delivery.AssignerAsk {
func RestoreAsk(key, node, module, fp, words string, now time.Time) asks.Ask {
return assignerAsk(0, oc.Ask{
ID: delivery.AskingID(key, 0),
ID: meshdeliveryclient.AskingID(key, 0),
Headline: fmt.Sprintf("%s failed its health check on %s", module, node),
Explanation: fmt.Sprintf("%s holds data, and its health check failed on %s at %s: %s. Setting it back to the version that ran healthy "+
"would lose the writes made since; leaving it stopped keeps the data as it is.", module, node, short(fp), words),
@@ -157,14 +159,14 @@ const (
// RecordAsk is the ask about a move of a module of policy record on node to the composition with the digest, judged by the
// controller to recreate a container, to move data, or not to be judgeable (ADR 0539 §14). The want names the composition only
// after the answer yes. The way the move was judged is in the ask.
func RecordAsk(key, node, module, fp string, need RecordNeed, now time.Time) delivery.AssignerAsk {
func RecordAsk(key, node, module, fp string, need RecordNeed, now time.Time) asks.Ask {
what := map[Judgement]string{
JudgedRecreates: "recreates a container",
JudgedMovesData: "moves the module's data",
JudgedCannot: "cannot be judged: " + need.Words,
}[need.Judgement]
return assignerAsk(0, oc.Ask{
ID: delivery.AskingID(key, 0),
ID: meshdeliveryclient.AskingID(key, 0),
Headline: fmt.Sprintf("%s waits for your yes on %s", module, node),
Explanation: fmt.Sprintf("%s is recorded: a person takes each of its moves. The next one on %s (version %s) %s. "+
"The node is told to apply it only after you say yes.", module, node, short(fp), what),
@@ -180,7 +182,7 @@ func RecordAsk(key, node, module, fp string, need RecordNeed, now time.Time) del
}
// NextBackupAsking is the number of the asking that follows the one numbered after (a negative number: none yet). Numbers 0 and 1
// give one id (delivery.AskingID), so the first asking is 0 and the next 2.
// give one id (meshdeliveryclient.AskingID), so the first asking is 0 and the next 2.
func NextBackupAsking(after int) int {
if after < 0 {
return 0
@@ -193,7 +195,7 @@ var backupDigest = regexp.MustCompile(`\(composition (sha256:[0-9a-f]{64})\)`)
// BackupAskDigest is the digest of the composition a backup ask was made about, from its explanation; empty for an ask that names
// none. An answer is taken for the composition of the ask it answers, and a new revision at the same fingerprint is asked again
// under a new number (ADR 0539 §16).
func BackupAskDigest(a delivery.AssignerAsk) string {
func BackupAskDigest(a asks.Ask) string {
ask, err := AskOf(a)
if err != nil {
return ""
@@ -213,9 +215,9 @@ func BackupAnswerWords(option string, asking int) string {
// BackupAsk is the ask about a backup of a module on node that failed before an apply (ADR 0539 §16): back up again, or go on
// without a backup. digest is the composition about to be applied and asking the number of the asking, 0 the first; words are the
// backup's own words, from the node's have.
func BackupAsk(key, node, module, fp, digest, words string, asking int, now time.Time) delivery.AssignerAsk {
func BackupAsk(key, node, module, fp, digest, words string, asking int, now time.Time) asks.Ask {
return assignerAsk(asking, oc.Ask{
ID: delivery.AskingID(key, asking),
ID: meshdeliveryclient.AskingID(key, asking),
Headline: fmt.Sprintf("The backup of %s failed on %s", module, node),
Explanation: fmt.Sprintf("The backup before applying %s at %s on %s failed: %s. %s keeps running what it has until you answer. (composition %s)",
module, short(fp), node, words, module, digest),
modules/mesh-delivery-assigner/internal/assign/asks_test.go
+16 -13
View File
@@ -3,6 +3,9 @@ package assign
import (
"fmt"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
"slices"
"strings"
"testing"
@@ -22,7 +25,7 @@ func TestTheHoldAskIsValid(t *testing.T) {
}
func theHoldAskIsValid(t *testing.T) {
key, err := delivery.HoldAskKey("mail", fNew)
key, err := asks.HoldAskKey("mail", fNew)
if err != nil {
t.Fatal(err)
}
@@ -40,13 +43,13 @@ func theHoldAskIsValid(t *testing.T) {
}
// 1906/H2: an ask written again after a lift is another asking, under an id of its own.
again := HoldAsk(key, "mail", fNew, "ace", "x", 7, now)
if again.Asking != 7 || mustAsk(t, again).ID != delivery.AskingID(key, 7) || mustAsk(t, again).ID == mustAsk(t, a).ID {
if again.Asking != 7 || mustAsk(t, again).ID != meshdeliveryclient.AskingID(key, 7) || mustAsk(t, again).ID == mustAsk(t, a).ID {
t.Errorf("the asking after a lift = %d %s", again.Asking, mustAsk(t, again).ID)
}
if _, err := EncodeAsk(key, again); err != nil {
t.Errorf("the ask asked again cannot be written: %v", err)
}
if _, err := delivery.ViewAnswerKey(key); err != nil {
if _, err := viewstate.ViewAnswerKey(key); err != nil {
t.Errorf("the view may not answer it: %v", err)
}
// The join (A1d ruling 1): mesh-sdk no longer refuses another id, no option or no ask in an assigner's ask; EncodeAsk does.
@@ -58,7 +61,7 @@ func theHoldAskIsValid(t *testing.T) {
ask := mustAsk(t, good)
ask.Options = nil
empty.Ask = assignerAsk(0, ask).Ask
for name, a := range map[string]delivery.AssignerAsk{"another id": other, "no option": empty, "no ask": {}} {
for name, a := range map[string]asks.Ask{"another id": other, "no option": empty, "no ask": {}} {
if _, err := EncodeAsk(key, a); err == nil {
t.Errorf("%s: written", name)
}
@@ -69,7 +72,7 @@ func theHoldAskIsValid(t *testing.T) {
// 1906/D3 (ADR 0539 §19): the ask about a data module held on a node is a valid ask of the router and of mesh-sdk's layout at
// its restore key, of a kind the view may answer; it offers restoring (losing the newer writes) and leaving it stopped.
func theRestoreAskIsValid(t *testing.T) {
key, err := delivery.RestoreAskKey("ace", "mail", fNew)
key, err := asks.RestoreAskKey("ace", "mail", fNew)
if err != nil {
t.Fatal(err)
}
@@ -81,7 +84,7 @@ func theRestoreAskIsValid(t *testing.T) {
if _, err := EncodeAsk(key, a); err != nil {
t.Errorf("the ask cannot be written: %v", err)
}
if _, err := delivery.ViewAnswerKey(key); err != nil {
if _, err := viewstate.ViewAnswerKey(key); err != nil {
t.Errorf("the view may not answer it: %v", err)
}
var ids []string
@@ -175,25 +178,25 @@ func saidConditionsAreFindings(t *testing.T) {
// the view, and name the move and the node.
func TestTheRecordAndBackupAsksAreValid(t *testing.T) {
now := time.Now()
rkey, err := delivery.RecordAskKey("ace", "mail", "sha256:"+strings.Repeat("b", 64))
rkey, err := asks.RecordAskKey("ace", "mail", "sha256:"+strings.Repeat("b", 64))
if err != nil {
t.Fatal(err)
}
r := RecordAsk(rkey, "ace", "mail", fNew, RecordNeed{Node: "ace", Digest: "sha256:" + strings.Repeat("b", 64), Judgement: JudgedCannot, Words: "the composition it runs is gone"}, now)
bkey, err := delivery.BackupAskKey("ace", "mail", fNew)
bkey, err := asks.BackupAskKey("ace", "mail", fNew)
if err != nil {
t.Fatal(err)
}
dgst := "sha256:" + strings.Repeat("b", 64)
b := BackupAsk(bkey, "ace", "mail", fNew, dgst, "no space left", 0, now)
for key, a := range map[string]delivery.AssignerAsk{rkey: r, bkey: b} {
for key, a := range map[string]asks.Ask{rkey: r, bkey: b} {
if err := occ.Check(mustAsk(t, a), now); err != nil {
t.Errorf("%s: the router refuses the ask: %v", key, err)
}
if _, err := EncodeAsk(key, a); err != nil {
t.Errorf("%s: cannot be written: %v", key, err)
}
if _, err := delivery.ViewAnswerKey(key); err != nil {
if _, err := viewstate.ViewAnswerKey(key); err != nil {
t.Errorf("%s: the view may not answer it: %v", key, err)
}
}
@@ -235,7 +238,7 @@ func TestTheRecordAndBackupAsksAreValid(t *testing.T) {
if _, err := EncodeAsk(rkey, a); err != nil {
t.Errorf("bus %v: the put's ask cannot be written: %v", bus, err)
}
viewed := delivery.ViewAnswerFits(delivery.ViewAnswer{Asking: mustAsk(t, a).ID, Option: OptionYes}, rkey, a)
viewed := viewstate.ViewAnswerFits(viewstate.ViewAnswer{Asking: mustAsk(t, a).ID, Option: OptionYes}, rkey, a)
yes := mustAsk(t, a).Options[0]
// mesh-sdk gives no refusal by level any more: the teller refuses an option a view cannot answer (its level is not
// acknowledge), as it did through the sdk before.
@@ -284,7 +287,7 @@ func TestTheNeitherConditionNamesModuleNodeAndRule(t *testing.T) {
}
// mustAsk is the router's ask an asks-state value carries.
func mustAsk(t *testing.T, a delivery.AssignerAsk) oc.Ask {
func mustAsk(t *testing.T, a asks.Ask) oc.Ask {
t.Helper()
ask, err := AskOf(a)
if err != nil {
@@ -293,7 +296,7 @@ func mustAsk(t *testing.T, a delivery.AssignerAsk) oc.Ask {
return ask
}
func expiresAt(t *testing.T, a delivery.AssignerAsk) time.Time {
func expiresAt(t *testing.T, a asks.Ask) time.Time {
t.Helper()
return ExpiresAt(mustAsk(t, a))
}
modules/mesh-delivery-assigner/internal/assign/put.go
+6 -4
View File
@@ -4,6 +4,8 @@ import (
"fmt"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"time"
@@ -120,7 +122,7 @@ func StepPut(w PutWorld) PutNext {
next := say(delivery.PutPerson, "waits for the operator's yes to putting %s at %s on %s (composition %s, judged %s)", w.Module, fp, w.Node, short(d.Digest), d.Judgement)
if p.Ask == "" {
next.Ask = true
next.Put.Ask, _ = delivery.RecordAskKey(w.Node, w.Module, d.Digest)
next.Put.Ask, _ = asks.RecordAskKey(w.Node, w.Module, d.Digest)
}
return next
}
@@ -221,7 +223,7 @@ func PutAct(node, module, fp, composed string) occ.Act {
// PutAsk is the ask about a put's move (ADR 0546 clauses 7 and 8), under the record ask's key of the composition so the
// teller relays it and its answer as any record ask's. For the bus its yes is of level approve and binds the act, so only a
// channel that proves who answered takes it (ADR 0259), never the view.
func PutAsk(key, node, module string, p delivery.Put, d ComposedDigest, bus bool, now time.Time) (delivery.AssignerAsk, error) {
func PutAsk(key, node, module string, p delivery.Put, d ComposedDigest, bus bool, now time.Time) (asks.Ask, error) {
yes := oc.Option{ID: OptionYes, Label: "Yes, put it", Does: "the node is told to apply " + module + " at " + short(p.FP), Level: occ.Acknowledge}
what := map[Judgement]string{
JudgedRecreates: "recreates a container",
@@ -235,7 +237,7 @@ func PutAsk(key, node, module string, p delivery.Put, d ComposedDigest, bus bool
if bus {
binds, err := occ.ActDigest(PutAct(node, module, p.FP, d.Digest))
if err != nil {
return delivery.AssignerAsk{}, err
return asks.Ask{}, err
}
yes.Level, yes.Binds = occ.Approve, binds
yes.Does = "the bus's streams are snapshotted, then " + node + " is told to apply the bus at " + short(p.FP)
@@ -243,7 +245,7 @@ func PutAsk(key, node, module string, p delivery.Put, d ComposedDigest, bus bool
expires = occ.FormatTime(now.Add(occ.ApproveLasts))
}
return assignerAsk(PutAsking(p), oc.Ask{
ID: delivery.AskingID(key, PutAsking(p)),
ID: meshdeliveryclient.AskingID(key, PutAsking(p)),
Headline: fmt.Sprintf("Put %s on %s?", module, node),
Explanation: explanation,
Options: []oc.Option{yes,
modules/mesh-delivery-assigner/internal/assign/seat.go
+5 -3
View File
@@ -2,6 +2,8 @@ package assign
import (
"fmt"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"maps"
"slices"
"strings"
@@ -194,10 +196,10 @@ func PlacementAsking(p mma.Proposal, at time.Time) int {
// PlacementAsk is the authorising ask about a proposal (§5): an Approve that binds its act, so only a channel that
// proves who answered takes it, with the brief of the verdict by node (judge.Brief) as its explanation.
func PlacementAsk(key string, p mma.Proposal, brief string, asking int, now time.Time) (delivery.AssignerAsk, error) {
func PlacementAsk(key string, p mma.Proposal, brief string, asking int, now time.Time) (asks.Ask, error) {
binds, err := occ.ActDigest(PlacementAct(p))
if err != nil {
return delivery.AssignerAsk{}, err
return asks.Ask{}, err
}
nodes := plc.Nodes(p)
headline := fmt.Sprintf("Change placement on %s?", strings.Join(nodes, ", "))
@@ -205,7 +207,7 @@ func PlacementAsk(key string, p mma.Proposal, brief string, asking int, now time
headline = fmt.Sprintf("Change placement on %d machines?", len(nodes))
}
return assignerAsk(asking, oc.Ask{
ID: delivery.AskingID(key, asking),
ID: meshdeliveryclient.AskingID(key, asking),
Headline: headline,
Explanation: brief,
Options: []oc.Option{
modules/mesh-delivery-assigner/internal/assign/wants_test.go
+2 -1
View File
@@ -4,6 +4,7 @@ import (
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"slices"
@@ -884,7 +885,7 @@ func stepsOfAPut(t *testing.T) {
{"snapshotted: the want, warranted", func() PutWorld {
w := bus()
w.Answer, w.Snapshot = OptionYes, "novox's restore point"
w.Put.Ask, _ = delivery.RecordAskKey("novox", "nats", d.Digest)
w.Put.Ask, _ = asks.RecordAskKey("novox", "nats", d.Digest)
return w
}, delivery.PutWanted, "held until released; the streams were snapshotted: novox's restore point", false, false, true},
{"a user-list revision of the bus needs nobody", func() PutWorld {