Assigner: bound answers and asks by pulse keys instead of a timer, ask the operator for a hand-back, follow a partial write through, and call no controller verb for the seed (ADR 0567 §4, §5, §7, §9, review round 2)
This commit is contained in:
@@ -22,6 +22,7 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/judge"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
@@ -71,7 +72,11 @@ func startBed(t *testing.T, leaveOut string) *bed {
|
||||
if name == leaveOut {
|
||||
continue
|
||||
}
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
|
||||
cfg := jetstream.KeyValueConfig{Bucket: name, History: 5}
|
||||
if name == delivery.PlacementPulsesBucket {
|
||||
cfg.History, cfg.LimitMarkerTTL = 1, time.Minute // a time to live per key, as the controller makes it (key-ttl)
|
||||
}
|
||||
kv, err := js.CreateKeyValue(ctx, cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -1082,6 +1087,9 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
|
||||
return rev
|
||||
}
|
||||
anchorRev, laptopRev := place("anchor", "web"), place("laptop", "shell")
|
||||
bound := judge.AnswerBound
|
||||
t.Cleanup(func() { judge.AnswerBound = bound })
|
||||
judge.AnswerBound = 2 * time.Second
|
||||
saysWhatItKeeps(t, b, b.start(t))
|
||||
verbs := newSeatTools(bedStore{b}, func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"ok":true}`), nil })
|
||||
propose := func() delivery.PlacementProposal {
|
||||
@@ -1177,4 +1185,16 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
|
||||
if e, _ := b.buckets[delivery.PlacementBucket].Get(b.ctx, key); e == nil || !strings.Contains(string(e.Value()), "web") {
|
||||
t.Error("a declined proposal wrote its key")
|
||||
}
|
||||
|
||||
// ADR 0567 §4: a node that does not answer fails the verdict once its pulse key's time to live ends, heard on the bus
|
||||
// with no timer of the assigner's, naming it.
|
||||
q, err = verbs.doPropose(map[string]any{"why": "again", "changes": `[{"op":"take-off","node":"laptop","module":"web"}]`}, "op")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
silent := q.(proposed).Proposal
|
||||
failed := proposalNow(silent.ID, "failed for a silent node", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalFailed })
|
||||
if failed.Verdict == nil || !strings.Contains(strings.Join(failed.Verdict.Fails, "; "), "laptop (offline) did not answer") {
|
||||
t.Errorf("the silent node's verdict = %+v", failed.Verdict)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
"os"
|
||||
"os/signal"
|
||||
"slices"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -74,11 +73,12 @@ const goBuildModule = "mesh-build-go"
|
||||
// (ADR 0567 §1), named after the assigner as every module state is.
|
||||
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket,
|
||||
delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket,
|
||||
delivery.PlacementCandidatesBucket}
|
||||
delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket}
|
||||
|
||||
// seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants
|
||||
// them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind).
|
||||
var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket}
|
||||
var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket,
|
||||
delivery.PlacementPulsesBucket}
|
||||
|
||||
// keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view
|
||||
// (and the node agent from release 6), its differences by the view, its compositions by the checker, its asks by the view
|
||||
@@ -88,6 +88,7 @@ var keptStates = []announce.State{
|
||||
{Name: delivery.PlacementState, Readable: true},
|
||||
{Name: delivery.PlacementProposalsState, Readable: true},
|
||||
{Name: delivery.PlacementCandidatesState, Readable: true},
|
||||
{Name: delivery.PlacementPulsesState, Readable: true},
|
||||
{Name: "asks", Readable: true},
|
||||
{Name: "compositions", Readable: true},
|
||||
{Name: "differences", Readable: true},
|
||||
@@ -177,6 +178,7 @@ var watched = []string{
|
||||
delivery.PlacementBucket,
|
||||
delivery.PlacementProposalsBucket,
|
||||
delivery.PlacementCandidatesBucket,
|
||||
delivery.PlacementPulsesBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.AssignerAsksBucket,
|
||||
@@ -317,7 +319,6 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
}
|
||||
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, said: map[string]bool{}, answeredAt: map[string]uint64{}}
|
||||
p.wakeAt = wakeAtVia(ctx, wake)
|
||||
p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
|
||||
if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
|
||||
p.acts = newPutActs(bus, wake, logger.Printf)
|
||||
@@ -347,31 +348,6 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
}
|
||||
|
||||
// wakeAtVia asks wake once at each time it is given, as a context's deadline (ADR 0437 §8: no timer), each time once.
|
||||
func wakeAtVia(ctx context.Context, wake func()) func(time.Time) {
|
||||
var mu sync.Mutex
|
||||
asked := map[int64]bool{}
|
||||
return func(at time.Time) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if asked[at.UnixNano()] {
|
||||
return
|
||||
}
|
||||
asked[at.UnixNano()] = true
|
||||
go func() {
|
||||
bound, cancel := context.WithDeadline(ctx, at)
|
||||
defer cancel()
|
||||
<-bound.Done()
|
||||
mu.Lock()
|
||||
delete(asked, at.UnixNano())
|
||||
mu.Unlock()
|
||||
if ctx.Err() == nil {
|
||||
wake()
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// PutsReopenEvery is how often, at most, a pass opens the puts' bucket again while it is not open.
|
||||
var PutsReopenEvery = time.Minute
|
||||
|
||||
|
||||
@@ -43,16 +43,10 @@ type passer struct {
|
||||
// nodeFailed is, by node and module, the words of the condition last raised for a failure that is the node's own (ADR 0557
|
||||
// §1, issue 1496): raised once per words, cleared once when it ends.
|
||||
nodeFailed map[string]string
|
||||
// wakeAt asks a pass at a time: the end of a proposal's answer bound (ADR 0567 §4), as a context's deadline (ADR 0437
|
||||
// §8: no timer). Nil asks none.
|
||||
wakeAt func(at time.Time)
|
||||
// The seat mesh-placement's holder (seat.go, ADR 0567): its calls of the controller (nil: none is made), how it says
|
||||
// the seat's events (nil: none is said), the placement keys it saw last (a key gone that it did not delete is a
|
||||
// hand-back), the proposals it said, and what Intended computed by proposal and base.
|
||||
// the seat's events (nil: none is said), the proposals it said, and what Intended computed by proposal and base.
|
||||
seat *seatActs
|
||||
publish publisher
|
||||
seen map[string]delivery.Placement
|
||||
seenRev map[string]uint64
|
||||
saidProposed map[string]bool
|
||||
intended map[string]map[string]map[string]bool
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -126,6 +127,16 @@ func (a *seatActs) record(warrant string) {
|
||||
}()
|
||||
}
|
||||
|
||||
// handAct records a hand-back in the controller's hand-act log (§7).
|
||||
func (a *seatActs) handAct(prop delivery.PlacementProposal) {
|
||||
go func() {
|
||||
if _, err := a.bus.Call(handActCall, map[string]any{"what": "mesh-placement hand-back of " + strings.Join(prop.Nodes(), ", "),
|
||||
"why": prop.Why, "cause": "asked by " + prop.By + ", approved by the operator"}); err != nil {
|
||||
a.logf("the hand-back %s is not recorded in the hand-act log: %v", prop.ID, err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// callAll makes the controller's calls in order in the background, each said in the journal and kept by its pair.
|
||||
func (a *seatActs) callAll(calls []assign.ControllerCall) {
|
||||
if len(calls) == 0 {
|
||||
@@ -225,16 +236,17 @@ func (p *passer) answersTo(id string) map[string]*delivery.CandidateAnswer {
|
||||
return out
|
||||
}
|
||||
|
||||
// placementSeat takes every open proposal one step further, and says the keys a hand-back deleted.
|
||||
// placementSeat takes every open proposal one step further.
|
||||
func (p *passer) placementSeat(ctx context.Context) {
|
||||
if p.kvs[delivery.PlacementProposalsBucket] == nil || p.kvs[delivery.PlacementBucket] == nil || p.kvs[delivery.PlacementCandidatesBucket] == nil {
|
||||
return
|
||||
for _, b := range seatBuckets {
|
||||
if p.kvs[b] == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
if p.saidProposed == nil {
|
||||
p.saidProposed, p.intended = map[string]bool{}, map[string]map[string]map[string]bool{}
|
||||
}
|
||||
current, revs := p.seatPlacements()
|
||||
p.handedBack(current, revs)
|
||||
ids := make([]string, 0)
|
||||
raws := p.w.read(delivery.PlacementProposalsBucket)
|
||||
for id := range raws {
|
||||
@@ -251,39 +263,46 @@ func (p *passer) placementSeat(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// handedBack says placement-changed for every key that went without this process deleting it: a hand-back (§7).
|
||||
func (p *passer) handedBack(current map[string]delivery.Placement, revs map[string]uint64) {
|
||||
if p.seen == nil {
|
||||
p.seen = map[string]delivery.Placement{}
|
||||
p.seenRev = map[string]uint64{}
|
||||
for n, pl := range current {
|
||||
p.seen[n], p.seenRev[n] = pl, revs[n]
|
||||
}
|
||||
// pulse writes a bound's key with its time to live (§4): its expiry, heard on the watch as the server's delete marker,
|
||||
// is the event that ends the wait. Written again over one that stands, so its time starts anew.
|
||||
func (p *passer) pulse(ctx context.Context, key string, ttl time.Duration) {
|
||||
kv := p.kvs[delivery.PlacementPulsesBucket]
|
||||
_ = kv.Delete(ctx, key)
|
||||
rev, err := kv.Create(ctx, key, []byte(`{}`), jetstream.KeyTTL(ttl))
|
||||
if err != nil {
|
||||
p.logf("the bound %s is not written, so its wait ends at once: %v", key, err)
|
||||
return
|
||||
}
|
||||
var gone []delivery.PlacementChange
|
||||
for node, before := range p.seen {
|
||||
if _, there := current[node]; there {
|
||||
continue
|
||||
p.w.putEntry(delivery.PlacementPulsesBucket, key, []byte(`{}`), rev, p.clock())
|
||||
}
|
||||
|
||||
// pulsing says whether a bound's key stands.
|
||||
func (p *passer) pulsing(key string) bool { return p.w.has(delivery.PlacementPulsesBucket, key) }
|
||||
|
||||
// deletePulses deletes every bound of a proposal.
|
||||
func (p *passer) deletePulses(ctx context.Context, id string) {
|
||||
for key := range p.w.read(delivery.PlacementPulsesBucket) {
|
||||
if strings.HasSuffix(key, "."+id) || strings.Contains(key, "."+id+".") {
|
||||
if err := p.kvs[delivery.PlacementPulsesBucket].Delete(ctx, key); err == nil {
|
||||
p.w.put(delivery.PlacementPulsesBucket, key, nil)
|
||||
}
|
||||
}
|
||||
c := delivery.ChangeOf(before, delivery.Placement{Node: node})
|
||||
c.Before = p.seenRev[node]
|
||||
gone = append(gone, c)
|
||||
delete(p.seen, node)
|
||||
delete(p.seenRev, node)
|
||||
}
|
||||
for n, pl := range current {
|
||||
p.seen[n], p.seenRev[n] = pl, revs[n]
|
||||
}
|
||||
|
||||
// ask raises (or raises again) a proposal's authorising ask and its bound, and answers the proposal asked.
|
||||
func (p *passer) ask(ctx context.Context, prop delivery.PlacementProposal, akey, brief string) (delivery.PlacementProposal, bool) {
|
||||
a, err := assign.PlacementAsk(akey, prop, brief, assign.PlacementAsking(prop, p.clock()), p.clock())
|
||||
if err != nil {
|
||||
p.logf("the ask about the proposal %s is not written: %v", prop.ID, err)
|
||||
return prop, false
|
||||
}
|
||||
if len(gone) == 0 || p.publish == nil {
|
||||
return
|
||||
}
|
||||
delivery.SortChanges(gone)
|
||||
ev := delivery.PlacementEvent{Proposal: delivery.PlacementProposal{ID: "hand-back", By: "the operator's hand act",
|
||||
Why: "placement handed back to the controller's compositions"}, Nodes: gone, Ended: "handed back"}
|
||||
if err := p.publish(delivery.PlacementChangedEvent, ev, fmt.Sprintf("hand-back.%d", p.clock().UnixNano())); err != nil {
|
||||
p.logf("the hand-back of placement is not said as an event: %v", err)
|
||||
p.put(ctx, delivery.AssignerAsksBucket, akey, func() ([]byte, error) { return delivery.EncodeAssignerAsk(akey, a) })
|
||||
if pk, err := delivery.PulseAskKey(prop.ID); err == nil {
|
||||
p.pulse(ctx, pk, a.Ask.Expires.Sub(p.clock()))
|
||||
}
|
||||
prop.State, prop.Ask = delivery.ProposalAsked, a.Ask.ID
|
||||
return prop, true
|
||||
}
|
||||
|
||||
// stepProposal takes one proposal one step further.
|
||||
@@ -297,21 +316,38 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos
|
||||
return
|
||||
}
|
||||
akey, _ := delivery.PlacementAskKey(prop.ID)
|
||||
again := !prop.Again.IsZero() && (prop.Verdict == nil || prop.Again.After(prop.Verdict.At))
|
||||
if prop.State == delivery.ProposalAsked {
|
||||
askPulse, _ := delivery.PulseAskKey(prop.ID)
|
||||
switch {
|
||||
case p.answeredWith(akey, assign.OptionYes):
|
||||
p.warranted(ctx, prop, rev, akey, current, revs)
|
||||
if prop.HandBack {
|
||||
p.handBack(ctx, prop, akey, revs)
|
||||
} else {
|
||||
p.warranted(ctx, prop, rev, akey, current, revs)
|
||||
}
|
||||
return
|
||||
case p.answeredWith(akey, assign.OptionNo):
|
||||
p.endProposal(ctx, prop, "declined by the operator")
|
||||
return
|
||||
}
|
||||
if a, err := delivery.DecodeAssignerAsk(akey, p.w.value(delivery.AssignerAsksBucket, akey)); err == nil &&
|
||||
!a.Ask.Expires.IsZero() && p.clock().After(a.Ask.Expires) && (prop.Again.IsZero() || prop.Again.Before(prop.Verdict.At)) {
|
||||
case !p.pulsing(askPulse) && !again:
|
||||
p.endProposal(ctx, prop, "its ask expired unanswered")
|
||||
return
|
||||
}
|
||||
}
|
||||
if prop.HandBack {
|
||||
// The operator's alone (§7): no node answers it; its ask is raised at once, and again on again.
|
||||
if prop.State != delivery.ProposalAsked || again {
|
||||
brief := fmt.Sprintf("%s (asked by %s): every node's placement key is deleted, %s, and each node returns to the "+
|
||||
"controller's compositions; the history is kept", prop.Why, prop.By, strings.Join(prop.Nodes(), ", "))
|
||||
now := p.clock().UTC()
|
||||
prop.Verdict = &delivery.PlacementVerdict{State: delivery.VerdictPass, At: now}
|
||||
if prop, ok := p.ask(ctx, prop, akey, brief); ok {
|
||||
p.writeProposal(ctx, prop, rev)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
if moved(prop, revs) {
|
||||
p.layAgain(ctx, prop, rev, current, revs, "a key it was laid over moved since")
|
||||
return
|
||||
@@ -329,11 +365,13 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos
|
||||
if written.After(prop.Written) {
|
||||
prop.Written = written
|
||||
}
|
||||
again := !prop.Again.IsZero() && (prop.Verdict == nil || prop.Again.After(prop.Verdict.At))
|
||||
v := judge.JudgeAnswers(prop, p.intendedOf(prop, current, cands), p.answersTo(prop.ID), candRevs, p.w.online(), p.clock())
|
||||
if v.State == delivery.VerdictWaits && p.wakeAt != nil {
|
||||
p.wakeAt(v.Until)
|
||||
pulsing := map[string]bool{}
|
||||
for _, node := range prop.Must {
|
||||
if k, err := delivery.PulseAnswerKey(prop.ID, node); err == nil {
|
||||
pulsing[node] = p.pulsing(k)
|
||||
}
|
||||
}
|
||||
v := judge.JudgeAnswers(prop, p.intendedOf(prop, current, cands), p.answersTo(prop.ID), candRevs, p.w.online(), pulsing, p.clock())
|
||||
if prop.Verdict != nil && sameVerdict(*prop.Verdict, v) && !again {
|
||||
return
|
||||
}
|
||||
@@ -345,14 +383,10 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos
|
||||
prop.State, prop.Ask = delivery.ProposalAnswering, ""
|
||||
case delivery.VerdictPass:
|
||||
if prop.State != delivery.ProposalAsked || again {
|
||||
asking := assign.PlacementAsking(prop, p.clock())
|
||||
a, err := assign.PlacementAsk(akey, prop, judge.Brief(prop, v), asking, p.clock())
|
||||
if err != nil {
|
||||
p.logf("the ask about the proposal %s is not written: %v", prop.ID, err)
|
||||
var ok bool
|
||||
if prop, ok = p.ask(ctx, prop, akey, judge.Brief(prop, v)); !ok {
|
||||
return
|
||||
}
|
||||
p.put(ctx, delivery.AssignerAsksBucket, akey, func() ([]byte, error) { return delivery.EncodeAssignerAsk(akey, a) })
|
||||
prop.State, prop.Ask = delivery.ProposalAsked, a.Ask.ID
|
||||
if again {
|
||||
prop.Words = "judged and asked again (again)"
|
||||
}
|
||||
@@ -454,6 +488,11 @@ func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal
|
||||
}
|
||||
p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return delivery.EncodePlacementCandidate(c) })
|
||||
}
|
||||
for _, node := range must {
|
||||
if k, err := delivery.PulseAnswerKey(prop.ID, node); err == nil {
|
||||
p.pulse(ctx, k, judge.AnswerBound)
|
||||
}
|
||||
}
|
||||
prop.Must = must
|
||||
if len(notes) > 0 {
|
||||
prop.Words = strings.TrimPrefix(prop.Words+"; "+strings.Join(notes, "; "), "; ")
|
||||
@@ -473,6 +512,7 @@ func (p *passer) layAgain(ctx context.Context, prop delivery.PlacementProposal,
|
||||
return
|
||||
}
|
||||
p.deleteCandidates(ctx, prop.ID)
|
||||
p.deletePulses(ctx, prop.ID)
|
||||
for _, node := range again.Nodes() {
|
||||
key, _ := delivery.PlacementCandidateKey(prop.ID, node)
|
||||
c := delivery.PlacementCandidate{Proposal: prop.ID, Placement: laid[node]}
|
||||
@@ -541,8 +581,11 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal,
|
||||
}
|
||||
kv := p.kvs[delivery.PlacementBucket]
|
||||
var changes []delivery.PlacementChange
|
||||
writtenLaid := map[string]*delivery.Placement{}
|
||||
after := maps.Clone(current)
|
||||
afterRevs := maps.Clone(revs)
|
||||
for _, w := range writes {
|
||||
var after uint64
|
||||
var rev2 uint64
|
||||
var err error
|
||||
switch {
|
||||
case w.Value == nil:
|
||||
@@ -550,47 +593,100 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal,
|
||||
default:
|
||||
raw, encErr := delivery.EncodePlacement(*w.Value)
|
||||
if encErr != nil {
|
||||
p.endProposal(ctx, prop, "its placement of "+w.Node+" cannot be written: "+encErr.Error())
|
||||
return
|
||||
}
|
||||
if w.Base == 0 {
|
||||
after, err = kv.Create(ctx, w.Key, raw)
|
||||
err = encErr
|
||||
} else if w.Base == 0 {
|
||||
rev2, err = kv.Create(ctx, w.Key, raw)
|
||||
} else {
|
||||
after, err = kv.Update(ctx, w.Key, raw, w.Base)
|
||||
rev2, err = kv.Update(ctx, w.Key, raw, w.Base)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
// Another writer came first: a hand-back. What was written stands, said; the rest is laid again.
|
||||
p.logf("placement %s is not written at revision %d (%v): the proposal %s is laid again", w.Key, w.Base, err, prop.ID)
|
||||
p.sayChanged(prop, answer, changes)
|
||||
p.layAgain(ctx, prop, rev, current, revs, "a key moved while it was written")
|
||||
// Another writer came first (a hand-back): what is written stands and is followed through; the rest is laid
|
||||
// again over the keys as this write left them.
|
||||
p.logf("placement %s is not written at revision %d (%v): the rest of the proposal %s is laid again", w.Key, w.Base, err, prop.ID)
|
||||
p.followWrite(prop, answer, changes, current, writtenLaid)
|
||||
if rest, ok := assign.Rest(prop, writtenLaid); ok {
|
||||
p.layAgain(ctx, rest, rev, after, afterRevs, "a key moved while it was written, after "+strings.Join(sortedNodes(writtenLaid), ", ")+" were")
|
||||
} else {
|
||||
p.endProposal(ctx, prop, "a key moved while it was written: "+err.Error())
|
||||
}
|
||||
return
|
||||
}
|
||||
before := current[w.Node]
|
||||
after2 := delivery.Placement{Node: w.Node}
|
||||
now := delivery.Placement{Node: w.Node}
|
||||
if w.Value != nil {
|
||||
after2 = *w.Value
|
||||
now = *w.Value
|
||||
after[w.Node] = *w.Value
|
||||
} else {
|
||||
delete(after, w.Node)
|
||||
}
|
||||
c := delivery.ChangeOf(before, after2)
|
||||
c.Before, c.After = w.Base, after
|
||||
afterRevs[w.Node] = rev2
|
||||
c := delivery.ChangeOf(before, now)
|
||||
c.Before, c.After = w.Base, rev2
|
||||
changes = append(changes, c)
|
||||
if p.seen != nil {
|
||||
if w.Value != nil {
|
||||
p.seen[w.Node], p.seenRev[w.Node] = *w.Value, after
|
||||
} else {
|
||||
delete(p.seen, w.Node)
|
||||
delete(p.seenRev, w.Node)
|
||||
}
|
||||
}
|
||||
writtenLaid[w.Node] = w.Value
|
||||
}
|
||||
p.seat.record(answer.Warrant)
|
||||
p.deleteCandidates(ctx, prop.ID)
|
||||
p.deletePulses(ctx, prop.ID)
|
||||
_ = p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID)
|
||||
p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil)
|
||||
p.sayChanged(prop, answer, changes)
|
||||
if p.seat != nil && !assign.Leased {
|
||||
p.seat.callAll(assign.ControllerCalls(current, laid, prop.ID, answer.Warrant))
|
||||
p.followWrite(prop, answer, changes, current, writtenLaid)
|
||||
}
|
||||
|
||||
// followWrite follows the keys a warrant wrote: the warrant recorded, placement-changed said, and, until the lease
|
||||
// write, the controller's assignments kept in step for them (§8). The seed calls no controller verb: it is what the
|
||||
// controller holds already (§9).
|
||||
func (p *passer) followWrite(prop delivery.PlacementProposal, answer delivery.Answer, changes []delivery.PlacementChange,
|
||||
before map[string]delivery.Placement, written map[string]*delivery.Placement) {
|
||||
if len(changes) == 0 {
|
||||
return
|
||||
}
|
||||
p.seat.record(answer.Warrant)
|
||||
p.sayChanged(prop, answer, changes)
|
||||
if p.seat != nil && !assign.Leased && !prop.Seed {
|
||||
p.seat.callAll(assign.ControllerCalls(before, written, prop.ID, answer.Warrant))
|
||||
}
|
||||
}
|
||||
|
||||
func sortedNodes(m map[string]*delivery.Placement) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for n := range m {
|
||||
out = append(out, n)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// handBack deletes every key on the operator's warrant (§7) at the revision the hand-back read, records it in the
|
||||
// controller's hand-act log and says placement-changed; every node returns to the controller's compositions. A key moved
|
||||
// since ends it, and the operator hands back again.
|
||||
func (p *passer) handBack(ctx context.Context, prop delivery.PlacementProposal, akey string, revs map[string]uint64) {
|
||||
if err := p.performs(akey, assign.PlacementAct(prop)); err != nil {
|
||||
p.endProposal(ctx, prop, "its warrant does not bind this hand-back, so nothing is deleted: "+err.Error())
|
||||
return
|
||||
}
|
||||
answer, _ := p.w.answer(akey)
|
||||
current, _ := p.seatPlacements()
|
||||
var changes []delivery.PlacementChange
|
||||
for _, node := range prop.Nodes() {
|
||||
key, _ := delivery.PlacementKey(node)
|
||||
if err := p.kvs[delivery.PlacementBucket].Delete(ctx, key, jetstream.LastRevision(prop.Base[node])); err != nil {
|
||||
p.sayChanged(prop, answer, changes)
|
||||
p.endProposal(ctx, prop, fmt.Sprintf("%s moved since the hand-back was asked (%v): hand back again", key, err))
|
||||
return
|
||||
}
|
||||
c := delivery.ChangeOf(current[node], delivery.Placement{Node: node})
|
||||
c.Before = prop.Base[node]
|
||||
changes = append(changes, c)
|
||||
}
|
||||
if p.seat != nil {
|
||||
p.seat.handAct(prop)
|
||||
}
|
||||
p.seat.record(answer.Warrant)
|
||||
_ = p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID)
|
||||
p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil)
|
||||
p.deletePulses(ctx, prop.ID)
|
||||
p.sayChanged(prop, answer, changes)
|
||||
}
|
||||
|
||||
// sayChanged says placement-changed for the keys written.
|
||||
@@ -638,6 +734,7 @@ func (p *passer) endProposal(ctx context.Context, prop delivery.PlacementProposa
|
||||
}
|
||||
}
|
||||
p.deleteCandidates(ctx, prop.ID)
|
||||
p.deletePulses(ctx, prop.ID)
|
||||
if err := p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID); err != nil && !errors.Is(err, context.Canceled) {
|
||||
p.logf("the proposal %s is not deleted, and is ended at the next pass: %v", prop.ID, err)
|
||||
return
|
||||
|
||||
@@ -311,8 +311,9 @@ func (s *seatTools) doAdopt(args map[string]any, by string) (any, error) {
|
||||
// handActCall records a hand act in the controller's hand-act log.
|
||||
const handActCall = "seat:" + catalogue.ControllerSeatName + ".hand-act"
|
||||
|
||||
// doHandBack is the verb hand-back (§7): every placement key deleted, the history kept, recorded as the operator's hand
|
||||
// act; every node returns to the controller's compositions. The assigner says placement-changed for each key it sees go.
|
||||
// doHandBack is the verb hand-back (§7): the operator's alone, so it is asked as a proposal is. It writes a hand-back
|
||||
// proposal naming every key at its revision; the assigner raises its authorising ask and, on its warrant, deletes the keys,
|
||||
// records the hand act and says placement-changed; every node returns to the controller's compositions.
|
||||
func (s *seatTools) doHandBack(args map[string]any, by string) (any, error) {
|
||||
if assign.Leased {
|
||||
return nil, errors.New("placement is the mesh's alone since the lease write: there is no controller to hand it back to")
|
||||
@@ -321,23 +322,28 @@ func (s *seatTools) doHandBack(args map[string]any, by string) (any, error) {
|
||||
if why == "" {
|
||||
return nil, errors.New("hand-back says why: it is recorded with the hand act")
|
||||
}
|
||||
keys, err := s.st.keys(delivery.PlacementState)
|
||||
if by == "" {
|
||||
return nil, errors.New("the runtime names no caller: a hand-back is made by someone")
|
||||
}
|
||||
_, revs, err := s.placements()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("placement cannot be listed: %w", err)
|
||||
return nil, err
|
||||
}
|
||||
sort.Strings(keys)
|
||||
if _, err := s.ask(handActCall, map[string]any{"what": "mesh-placement hand-back of " + strings.Join(keys, ", "),
|
||||
"why": why, "cause": "asked by " + by}); err != nil {
|
||||
return nil, fmt.Errorf("the hand-back is not recorded as a hand act, so nothing is deleted: %w", err)
|
||||
if len(revs) == 0 {
|
||||
return nil, errors.New("there is no placement key to hand back")
|
||||
}
|
||||
var deleted []string
|
||||
for _, key := range keys {
|
||||
if err := s.st.del(delivery.PlacementState, key); err != nil {
|
||||
return map[string]any{"deleted": deleted}, fmt.Errorf("%s is not deleted (%v); the others before it are: hand-back again", key, err)
|
||||
}
|
||||
deleted = append(deleted, key)
|
||||
now := s.now()
|
||||
p := delivery.PlacementProposal{ID: delivery.NewProposalID(now), By: by, Why: why, At: now.UTC(), Written: now.UTC(),
|
||||
HandBack: true, Base: revs, State: delivery.ProposalAnswering}
|
||||
raw, err := delivery.EncodePlacementProposal(p)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"deleted": deleted, "follow": "each node returns to the controller's compositions as it hears its key go"}, nil
|
||||
if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil {
|
||||
return nil, fmt.Errorf("the hand-back is not written: %w", err)
|
||||
}
|
||||
return proposed{Proposal: p, Follow: "the operator is asked to approve it; on the warrant every key is deleted and each " +
|
||||
"node returns to the controller's compositions"}, nil
|
||||
}
|
||||
|
||||
// placementAnswer is what the verb placement answers.
|
||||
|
||||
@@ -2,7 +2,6 @@ package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
@@ -137,14 +136,11 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// hand-back: recorded as the operator's hand act, then every key deleted; refused when the hand act is not recorded.
|
||||
failing := newSeatTools(st, func(string, any) (json.RawMessage, error) { return nil, errors.New("no controller") })
|
||||
if _, err := failing.doHandBack(map[string]any{"why": "back"}, "op"); err == nil || len(st[delivery.PlacementState].values) != 1 {
|
||||
t.Errorf("a hand-back not recorded deleted keys: %v", err)
|
||||
}
|
||||
asked = nil
|
||||
if _, err := s.doHandBack(map[string]any{"why": "back"}, "op"); err != nil || !slices.Equal(asked, []string{handActCall}) || len(st[delivery.PlacementState].values) != 0 {
|
||||
t.Errorf("hand-back: %v, asked %v", err, asked)
|
||||
// hand-back: the operator's alone, written as a hand-back proposal naming every key at its revision, for the assigner
|
||||
// to ask; nothing is deleted by the verb.
|
||||
got, err = s.doHandBack(map[string]any{"why": "back"}, "op")
|
||||
if hb := got.(proposed).Proposal; err != nil || !hb.HandBack || hb.Base["anchor"] != uint64(len(raw)) || len(st[delivery.PlacementState].values) != 1 {
|
||||
t.Errorf("hand-back = %+v, %v", got, err)
|
||||
}
|
||||
if tools := s.tools(); len(tools) != 7 || tools[2].Name != catalogue.PlacementVerbPropose || tools[2].Run == nil || tools[2].Input == nil {
|
||||
t.Errorf("the seat's verbs as served: %+v", tools)
|
||||
|
||||
@@ -240,18 +240,14 @@ func ExportOf(raw []byte) (map[string]delivery.Placement, error) {
|
||||
if a.OK != nil && !*a.OK {
|
||||
return nil, fmt.Errorf("the controller's placement-export refused: %s", strings.TrimSpace(a.Output))
|
||||
}
|
||||
data := []byte(a.Answer)
|
||||
if len(data) == 0 || string(data) == "null" {
|
||||
out := a.Output
|
||||
start, end := strings.Index(out, "{"), strings.LastIndex(out, "}")
|
||||
if start < 0 || end < start {
|
||||
return nil, fmt.Errorf("the controller's placement-export printed no placements: %q", out)
|
||||
}
|
||||
data = []byte(out[start : end+1])
|
||||
}
|
||||
var byNode map[string]json.RawMessage
|
||||
if err := json.Unmarshal(data, &byNode); err != nil {
|
||||
return nil, fmt.Errorf("the controller's placement-export is no placement by node: %w", err)
|
||||
if len(a.Answer) > 0 && string(a.Answer) != "null" {
|
||||
if err := json.Unmarshal(a.Answer, &byNode); err != nil {
|
||||
return nil, fmt.Errorf("the controller's placement-export is no placement by node: %w", err)
|
||||
}
|
||||
} else if err := json.NewDecoder(strings.NewReader(a.Output)).Decode(&byNode); err != nil {
|
||||
// The command prints its JSON first; what its standard error said after it is left.
|
||||
return nil, fmt.Errorf("the controller's placement-export printed no placement by node: %w", err)
|
||||
}
|
||||
out := map[string]delivery.Placement{}
|
||||
for node, rawP := range byNode {
|
||||
@@ -279,3 +275,22 @@ func SeedChanges(export map[string]delivery.Placement) []delivery.PlacementPropo
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Rest is what a proposal leaves to write once the keys of written are: its changes of the other nodes, to be laid again
|
||||
// over the keys as the write left them (§5: a key moved while it was written). False when nothing is left.
|
||||
func Rest(p delivery.PlacementProposal, written map[string]*delivery.Placement) (delivery.PlacementProposal, bool) {
|
||||
rest := p
|
||||
rest.Changes, rest.Base = nil, map[string]uint64{}
|
||||
for _, c := range p.Changes {
|
||||
if _, done := written[c.Node]; !done {
|
||||
rest.Changes = append(rest.Changes, c)
|
||||
}
|
||||
}
|
||||
for node, base := range p.Base {
|
||||
if _, done := written[node]; !done {
|
||||
rest.Base[node] = base
|
||||
}
|
||||
}
|
||||
rest.Must, rest.Verdict, rest.Ask, rest.State = nil, nil, "", delivery.ProposalAnswering
|
||||
return rest, len(rest.Changes) > 0
|
||||
}
|
||||
|
||||
@@ -39,6 +39,9 @@ func TestThePlacementSeatDecides(t *testing.T) {
|
||||
}{
|
||||
"a port before the lease": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int{"80": 8080}}), "lease write"},
|
||||
"a holding": {in(delivery.PlacementProposalChange{Op: delivery.OpHold, Node: "anchor", Module: "store", Claim: "mesh-store", Scope: "mesh"}), "lease write"},
|
||||
"a binding": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "web", Bindings: map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}}}), "lease write"},
|
||||
"a config place": {in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}}), "lease write"},
|
||||
"a config access": {in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", MeshWide: map[string]any{"accesses": []any{}}}), "lease write"},
|
||||
"a module not held": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "ghost"}), "the indexer holds no module ghost"},
|
||||
"a node with no record": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "stranger", Module: "web"}), "stranger is a node with no record"},
|
||||
"nothing changes": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "anchor", Module: "web"}), "leave every placement as it is"},
|
||||
@@ -52,6 +55,33 @@ func TestThePlacementSeatDecides(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A module's config values may be proposed before the lease write (§13); once the lease is written every change is
|
||||
// the state's alone, and none waits.
|
||||
if _, _, f := Propose(in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", Values: map[string]any{"puid": 1}})); len(f) != 0 {
|
||||
t.Errorf("config values before the lease write: %v", f)
|
||||
}
|
||||
Leased = true
|
||||
for _, c := range []delivery.PlacementProposalChange{
|
||||
{Op: delivery.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int{"80": 8080}, Bindings: map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}}},
|
||||
{Op: delivery.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}},
|
||||
{Op: delivery.OpDecide, Node: "laptop", Field: "site", Value: json.RawMessage(`"attic"`)},
|
||||
{Op: delivery.OpHold, Node: "anchor", Module: "store", Claim: "node-shell", Scope: "node"},
|
||||
} {
|
||||
if _, _, f := Propose(in(c)); len(f) != 0 {
|
||||
t.Errorf("after the lease write %s is refused: %v", c.Op, f)
|
||||
}
|
||||
}
|
||||
Leased = false
|
||||
|
||||
// A key moved while the warrant wrote: what is left to write is the other nodes' changes, to be laid and asked again.
|
||||
if rest, ok := Rest(p, map[string]*delivery.Placement{"anchor": laid["anchor"]}); !ok || len(rest.Changes) != 1 || rest.Changes[0].Node != "laptop" ||
|
||||
len(rest.Base) != 1 || rest.State != delivery.ProposalAnswering || rest.Must != nil {
|
||||
t.Errorf("the rest = %+v, %v", rest, ok)
|
||||
}
|
||||
if _, ok := Rest(p, laid); ok {
|
||||
t.Error("a proposal written whole leaves something to write")
|
||||
}
|
||||
|
||||
writes, err := Writes(p, laid)
|
||||
if err != nil || len(writes) != 2 || writes[0].Key != "placement.anchor" || writes[0].Base != 4 || writes[1].Base != 9 {
|
||||
t.Fatalf("writes = %+v, %v", writes, err)
|
||||
|
||||
@@ -44,7 +44,7 @@ func TestManifest(t *testing.T) {
|
||||
states[s.Name] = s.History
|
||||
}
|
||||
if want := map[string]int{"asks": 5, "compositions": 5, "differences": 64, "puts": 20, "wants": 20,
|
||||
"placement": 64, "proposals": 5, "candidates": 5}; !reflect.DeepEqual(states, want) {
|
||||
"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0}; !reflect.DeepEqual(states, want) {
|
||||
t.Errorf("state = %v, want %v", states, want)
|
||||
}
|
||||
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
|
||||
|
||||
@@ -59,6 +59,7 @@
|
||||
{ "name": "placement", "history": 64 },
|
||||
{ "name": "proposals", "history": 5 },
|
||||
{ "name": "candidates", "history": 5 },
|
||||
{ "name": "pulses", "key-ttl": true },
|
||||
{ "name": "asks", "history": 5 },
|
||||
{ "name": "compositions", "history": 5 },
|
||||
{ "name": "differences", "history": 64 },
|
||||
|
||||
Reference in New Issue
Block a user