diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go new file mode 100644 index 0000000..5f4d55c --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go @@ -0,0 +1,526 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-assigner/internal/assign" +) + +// The seat mesh-placement's seven verbs (novox/hq ADR 0567 §10), served through the tool runner beside the put tools: a +// process never serves (ADR 0436). They read every state the seat keeps and write what a person asks of it: a proposal +// and its candidates (propose, adopt), a proposal marked to end (withdraw) or to be judged again (again), and the keys +// deleted (hand-back). Everything that follows a proposal — composing it, judging the answers, the ask, the write at the +// warrant and the seat's events on its stream — is the assigner's process's (seat.go), started by these writes. + +// seatStore is what the seat's verbs need of the states: the runtime's, or a test's. +type seatStore interface { + store + del(state, key string) error + history(state, key string) ([]stdio.StateRevision, error) +} + +func (runtimeStore) del(state, key string) error { return stdio.State(state).Delete(key) } + +func (runtimeStore) history(state, key string) ([]stdio.StateRevision, error) { + return stdio.State(state).History(key) +} + +// The states the verbs read, as the runtime names them: the seat's own by their local names, the others' +// `.`. +const ( + modulesState = delivery.IndexerModule + ".modules" + haveState = "node-engine.have" +) + +// seatTools serves the seat's verbs over a store, asking the controller through ask (stdio.Ask, or a test's). +type seatTools struct { + st seatStore + ask func(key string, body any) (json.RawMessage, error) + now func() time.Time +} + +func newSeatTools(st seatStore, ask func(string, any) (json.RawMessage, error)) *seatTools { + return &seatTools{st: st, ask: ask, now: time.Now} +} + +// keyed is one placement key as a verb answers it. +type keyed struct { + Placement delivery.Placement `json:"placement"` + Revision uint64 `json:"revision"` +} + +// placements reads every node's placement and its key's revision. +func (s *seatTools) placements() (map[string]delivery.Placement, map[string]uint64, error) { + keys, err := s.st.keys(delivery.PlacementState) + if err != nil { + return nil, nil, fmt.Errorf("placement cannot be listed: %w", err) + } + out, revs := map[string]delivery.Placement{}, map[string]uint64{} + for _, key := range keys { + node, err := delivery.ParsePlacementKey(key) + if err != nil { + continue + } + raw, rev, found, err := s.st.get(delivery.PlacementState, key) + if err != nil { + return nil, nil, fmt.Errorf("placement of %s cannot be read: %w", node, err) + } + if !found { + continue + } + p, err := delivery.DecodePlacement(raw) + if err != nil { + return nil, nil, fmt.Errorf("placement of %s cannot be read: %w", node, err) + } + out[node], revs[node] = p, rev + } + return out, revs, nil +} + +// lays is what a proposal is checked against: the modules the indexer holds and the nodes with a record. +func (s *seatTools) lays() (func(string) bool, func(string) bool, error) { + mods, err := s.st.keys(modulesState) + if err != nil { + return nil, nil, fmt.Errorf("the indexer's modules cannot be listed: %w", err) + } + held := map[string]bool{} + for _, k := range mods { + if name, err := delivery.ParseModulesKey(k); err == nil { + held[name] = true + } + } + haves, err := s.st.keys(haveState) + if err != nil { + return nil, nil, fmt.Errorf("the nodes' records cannot be listed: %w", err) + } + recorded := map[string]bool{} + for _, k := range haves { + node, _, isNode, err := delivery.ParseHaveKey(k) + if err != nil || !isNode { + continue + } + if raw, _, found, err := s.st.get(haveState, k); err == nil && found { + if h, err := delivery.DecodeHaveNode(raw); err == nil && h.Record != nil { + recorded[node] = true + } + } + } + return func(m string) bool { return held[m] }, func(n string) bool { return recorded[n] }, nil +} + +// proposal reads one open proposal. +func (s *seatTools) proposal(id string) (delivery.PlacementProposal, error) { + if err := delivery.CheckProposalID(id); err != nil { + return delivery.PlacementProposal{}, err + } + raw, _, found, err := s.st.get(delivery.PlacementProposalsState, id) + if err != nil { + return delivery.PlacementProposal{}, fmt.Errorf("the proposal %s cannot be read: %w", id, err) + } + if !found { + return delivery.PlacementProposal{}, fmt.Errorf("no proposal %s is open: mesh-placement.proposals lists those that are", id) + } + return delivery.DecodePlacementProposal(raw) +} + +// write writes a proposal laid by assign.Propose: each candidate, then the proposal, whose write the assigner's process +// takes as the event to compose and judge it. +func (s *seatTools) write(p delivery.PlacementProposal, laid map[string]*delivery.Placement) error { + p.Written = s.now().UTC() + for _, node := range p.Nodes() { + key, err := delivery.PlacementCandidateKey(p.ID, node) + if err != nil { + return err + } + raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: p.ID, Placement: laid[node]}) + if err != nil { + return err + } + if err := s.st.put(delivery.PlacementCandidatesState, key, raw); err != nil { + return fmt.Errorf("the candidate %s is not written: %w", key, err) + } + } + raw, err := delivery.EncodePlacementProposal(p) + if err != nil { + return err + } + if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil { + return fmt.Errorf("the proposal %s is not written: %w", p.ID, err) + } + return nil +} + +// proposed is what propose and adopt answer. +type proposed struct { + Proposal delivery.PlacementProposal `json:"proposal"` + Follow string `json:"follow"` +} + +func followProposal(id string) string { + return fmt.Sprintf("the nodes it touches answer it; mesh-placement.proposals with id %q says where it stands, and the "+ + "operator is asked to approve it once its verdict passes", id) +} + +// changesOf reads propose's changes: a JSON array, given as one or as its text. +func changesOf(v any) ([]delivery.PlacementProposalChange, error) { + var raw []byte + switch c := v.(type) { + case string: + raw = []byte(c) + case nil: + return nil, errors.New("propose names its changes") + default: + var err error + if raw, err = json.Marshal(c); err != nil { + return nil, err + } + } + var out []delivery.PlacementProposalChange + dec := json.NewDecoder(strings.NewReader(string(raw))) + dec.DisallowUnknownFields() + if err := dec.Decode(&out); err != nil { + return nil, fmt.Errorf("the changes are no list of changes: %w", err) + } + return out, nil +} + +// doPropose is the verb propose (§2). +func (s *seatTools) doPropose(args map[string]any, by string) (any, error) { + changes, err := changesOf(args["changes"]) + if err != nil { + return nil, err + } + for _, c := range changes { + if c.Op == delivery.OpSeed { + return nil, errors.New("the seed is adopt's alone: propose takes place, take-off, pin, unpin, hold, unhold, decide and config") + } + } + current, revs, err := s.placements() + if err != nil { + return nil, err + } + held, recorded, err := s.lays() + if err != nil { + return nil, err + } + now := s.now() + p, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: by, Why: arg(args, "why"), At: now, + Changes: changes, Current: current, Revisions: revs, Held: held, Recorded: recorded}) + if len(findings) > 0 { + return nil, fmt.Errorf("the proposal is refused: %s", strings.Join(findings, "; ")) + } + if err := s.write(p, laid); err != nil { + return nil, err + } + return proposed{Proposal: p, Follow: followProposal(p.ID)}, nil +} + +// doWithdraw is the verb withdraw (§2): the proposal is marked, and the assigner deletes it and its candidates and says +// placement-withdrawn carrying it whole. +func (s *seatTools) doWithdraw(args map[string]any, by string) (any, error) { + why := arg(args, "why") + if why == "" { + return nil, errors.New("withdraw says why: it is kept in the event that ends the proposal") + } + p, err := s.proposal(arg(args, "id")) + if err != nil { + return nil, err + } + p.Withdrawn = fmt.Sprintf("withdrawn by %s: %s", by, why) + raw, err := delivery.EncodePlacementProposal(p) + if err != nil { + return nil, err + } + if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil { + return nil, fmt.Errorf("the proposal %s is not withdrawn: %w", p.ID, err) + } + return map[string]any{"proposal": p.ID, "withdrawn": p.Withdrawn}, nil +} + +// doAgain is the verb again (§10): the proposal is marked, and the assigner judges it now and raises its ask again +// when its verdict passes. +func (s *seatTools) doAgain(args map[string]any) (any, error) { + p, err := s.proposal(arg(args, "id")) + if err != nil { + return nil, err + } + p.Again = s.now().UTC() + raw, err := delivery.EncodePlacementProposal(p) + if err != nil { + return nil, err + } + if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil { + return nil, fmt.Errorf("the proposal %s is not asked again: %w", p.ID, err) + } + return map[string]any{"proposal": p.ID, "again": p.Again, "follow": followProposal(p.ID)}, nil +} + +// exportCall is the controller's verb the seed is read from, by request and reply on its seat (§7). +const exportCall = "seat:" + catalogue.ControllerSeatName + ".placement-export" + +// doAdopt is the verb adopt (§7): the controller's export proposed whole as the seed. +func (s *seatTools) doAdopt(args map[string]any, by string) (any, error) { + current, _, err := s.placements() + if err != nil { + return nil, err + } + if len(current) > 0 { + return nil, fmt.Errorf("placement is taken over already (%d keys): propose a change instead, or hand it back first", len(current)) + } + if ids, err := s.st.keys(delivery.PlacementProposalsState); err == nil { + for _, id := range ids { + if p, err := s.proposal(id); err == nil && p.Seed { + return nil, fmt.Errorf("the seed %s is open: withdraw it first, or answer its ask", id) + } + } + } + raw, err := s.ask(exportCall, map[string]any{}) + if err != nil { + return nil, fmt.Errorf("the controller's placement-export did not answer: %w", err) + } + export, err := assign.ExportOf(raw) + if err != nil { + return nil, err + } + held, recorded, err := s.lays() + if err != nil { + return nil, err + } + now := s.now() + p, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: by, Why: arg(args, "why"), At: now, + Changes: assign.SeedChanges(export), Current: map[string]delivery.Placement{}, Revisions: map[string]uint64{}, + Held: held, Recorded: recorded, Seed: true, ExportDigest: delivery.ExportDigest(export)}) + if len(findings) > 0 { + return nil, fmt.Errorf("the seed is refused: %s", strings.Join(findings, "; ")) + } + if err := s.write(p, laid); err != nil { + return nil, err + } + return proposed{Proposal: p, Follow: followProposal(p.ID)}, nil +} + +// 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. +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") + } + why := arg(args, "why") + 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 err != nil { + return nil, fmt.Errorf("placement cannot be listed: %w", 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) + } + 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) + } + return map[string]any{"deleted": deleted, "follow": "each node returns to the controller's compositions as it hears its key go"}, nil +} + +// placementAnswer is what the verb placement answers. +type placementAnswer struct { + Keys map[string]keyed `json:"keys,omitempty"` + Nodes []string `json:"nodes,omitempty"` + Count int `json:"count"` + History int `json:"history-per-key"` + Changes []historyEntry `json:"changes,omitempty"` + Older string `json:"older,omitempty"` +} + +// historyEntry is one change of a key in the state's revisions. +type historyEntry struct { + Node string `json:"node"` + Revision uint64 `json:"revision"` + At time.Time `json:"at"` + Change delivery.PlacementChange `json:"change"` + Deleted bool `json:"deleted,omitempty"` +} + +// doPlacement is the verb placement (§10, §6): one node's key, one module's nodes, or every key; with history, the +// changes in order from the revisions the state keeps. +func (s *seatTools) doPlacement(args map[string]any) (any, error) { + all, revs, err := s.placements() + if err != nil { + return nil, err + } + node, module := arg(args, "node"), arg(args, "module") + out := placementAnswer{Keys: map[string]keyed{}, Count: len(all), History: delivery.PlacementHistory} + for n, p := range all { + if node != "" && n != node { + continue + } + if module != "" && !strings.Contains(" "+strings.Join(p.Assigned(), " ")+" ", " "+module+" ") { + continue + } + out.Keys[n] = keyed{Placement: p, Revision: revs[n]} + out.Nodes = append(out.Nodes, n) + } + sort.Strings(out.Nodes) + if node != "" && len(out.Keys) == 0 && arg(args, "history") != "true" { + return nil, fmt.Errorf("%s has no placement key", node) + } + if arg(args, "history") != "true" { + return out, nil + } + nodes := out.Nodes + if node != "" { + nodes = []string{node} + } else if module == "" { + nodes = nil + for n := range all { + nodes = append(nodes, n) + } + sort.Strings(nodes) + } + for _, n := range nodes { + key, err := delivery.PlacementKey(n) + if err != nil { + return nil, err + } + revisions, err := s.st.history(delivery.PlacementState, key) + if err != nil { + return nil, fmt.Errorf("the history of %s cannot be read: %w", key, err) + } + var before delivery.Placement + for _, r := range revisions { + e := historyEntry{Node: n, Revision: r.Revision, At: r.At} + after := delivery.Placement{} + if r.Op == "" || r.Op == "put" || r.Op == "PUT" { + if p, err := delivery.DecodePlacement(r.Value); err == nil { + after = p + } + } else { + e.Deleted = true + } + e.Change = delivery.ChangeOf(before, after) + e.Change.Before, e.Change.After = 0, r.Revision + before = after + if module != "" && !strings.Contains(" "+strings.Join(append(append(e.Change.Added, e.Change.Removed...), e.Change.Changed...), " ")+" ", " "+module+" ") { + continue + } + out.Changes = append(out.Changes, e) + } + } + out.Older = fmt.Sprintf("the state keeps %d revisions of each key; an older change is in its placement-changed event on "+ + "the assigner's stream %s, kept %d days", delivery.PlacementHistory, delivery.PlacementStream, delivery.PlacementStreamDays) + return out, nil +} + +// openProposal is one open proposal as the verb proposals answers it: the proposal (its changes, base revisions, +// verdict with the differences and the answers it judged, its ask), its candidates by node with their revisions, and +// every node's answer now. +type openProposal struct { + Proposal delivery.PlacementProposal `json:"proposal"` + Candidates map[string]uint64 `json:"candidates"` + Answers map[string]delivery.CandidateAnswer `json:"answers,omitempty"` +} + +// doProposals is the verb proposals (§10). +func (s *seatTools) doProposals(args map[string]any) (any, error) { + ids, err := s.st.keys(delivery.PlacementProposalsState) + if err != nil { + return nil, fmt.Errorf("the proposals cannot be listed: %w", err) + } + cands, err := s.st.keys(delivery.PlacementCandidatesState) + if err != nil { + return nil, fmt.Errorf("the candidates cannot be listed: %w", err) + } + haves, err := s.st.keys(haveState) + if err != nil { + return nil, fmt.Errorf("the nodes' haves cannot be listed: %w", err) + } + answers := map[string]map[string]delivery.CandidateAnswer{} // by proposal, then node + for _, k := range haves { + node, _, isNode, err := delivery.ParseHaveKey(k) + if err != nil || !isNode { + continue + } + raw, _, found, err := s.st.get(haveState, k) + if err != nil || !found { + continue + } + h, err := delivery.DecodeHaveNode(raw) + if err != nil { + continue + } + for id, a := range h.Candidates { + if answers[id] == nil { + answers[id] = map[string]delivery.CandidateAnswer{} + } + answers[id][node] = a + } + } + sort.Strings(ids) + out := []openProposal{} + for _, id := range ids { + if want := arg(args, "id"); want != "" && want != id { + continue + } + p, err := s.proposal(id) + if err != nil { + return nil, err + } + o := openProposal{Proposal: p, Candidates: map[string]uint64{}, Answers: answers[id]} + for _, k := range cands { + pid, node, err := delivery.ParsePlacementCandidateKey(k) + if err != nil || pid != id { + continue + } + if _, rev, found, err := s.st.get(delivery.PlacementCandidatesState, k); err == nil && found { + o.Candidates[node] = rev + } + } + out = append(out, o) + } + if arg(args, "id") != "" && len(out) == 0 { + return nil, fmt.Errorf("no proposal %s is open", arg(args, "id")) + } + return map[string]any{"proposals": out}, nil +} + +// tools are the seat's verbs as the runtime serves them, under the seat's verb names, with the seat's own schemas. +func (s *seatTools) tools() []stdio.Tool { + seat, _ := catalogue.SeatNamed(catalogue.PlacementSeat) + input := map[string]map[string]any{} + description := map[string]string{} + for _, v := range seat.Serves { + input[v.Name], description[v.Name] = v.Input, v.Description + } + run := map[string]func(map[string]any) (any, error){ + catalogue.PlacementVerbPlacement: s.doPlacement, + catalogue.PlacementVerbProposals: s.doProposals, + catalogue.PlacementVerbPropose: func(a map[string]any) (any, error) { return s.doPropose(a, stdio.CallerOf(a)) }, + catalogue.PlacementVerbWithdraw: func(a map[string]any) (any, error) { return s.doWithdraw(a, stdio.CallerOf(a)) }, + catalogue.PlacementVerbAgain: s.doAgain, + catalogue.PlacementVerbAdopt: func(a map[string]any) (any, error) { return s.doAdopt(a, stdio.CallerOf(a)) }, + catalogue.PlacementVerbHandBack: func(a map[string]any) (any, error) { return s.doHandBack(a, stdio.CallerOf(a)) }, + } + var out []stdio.Tool + for _, v := range seat.Serves { + out = append(out, stdio.Tool{Name: v.Name, Description: description[v.Name], Input: input[v.Name], Run: run[v.Name]}) + } + return out +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go new file mode 100644 index 0000000..fdb459f --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -0,0 +1,161 @@ +package main + +import ( + "encoding/json" + "errors" + "os" + "slices" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func (s memStore) del(state, key string) error { + delete(s[state].values, key) + return nil +} + +func (s memStore) history(state, key string) ([]stdio.StateRevision, error) { + v, ok := s[state].values[key] + if !ok { + return nil, nil + } + return []stdio.StateRevision{{Key: key, Value: v, Revision: uint64(len(v)), Op: "put"}}, nil +} + +// ADR 0567 §2, §7, §10: the seat's seven verbs over the states. propose lays a proposal and writes it with its candidates, +// or refuses naming every finding and writes nothing; withdraw and again mark it for the holder; adopt proposes the +// controller's export whole as the seed, once; hand-back records the hand act and deletes every key; placement and +// proposals read every state the seat keeps, and history names the stream for what the state no longer keeps. +func TestTheSeatsVerbs(t *testing.T) { + anchor := delivery.Placement{V: delivery.Version, Node: "anchor", Modules: []delivery.PlacedModule{{Module: "web"}}} + st := memStore{} + for _, state := range []string{delivery.PlacementState, delivery.PlacementProposalsState, delivery.PlacementCandidatesState, modulesState, haveState} { + st[state] = &memKV{values: map[string][]byte{}} + } + for _, m := range []string{"web", "shell"} { + st[modulesState].values[m] = []byte("{}") + } + for _, n := range []string{"anchor", "laptop"} { + raw, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: n}}) + st[haveState].values[n+".node"] = raw + } + export, _ := json.Marshal(map[string]any{"anchor": anchor}) + var asked []string + s := newSeatTools(st, func(key string, body any) (json.RawMessage, error) { + asked = append(asked, key) + if key == exportCall { + out, _ := json.Marshal(map[string]any{"ok": true, "output": string(export)}) + return out, nil + } + return json.RawMessage(`{"ok":true}`), nil + }) + s.now = func() time.Time { return time.Date(2026, 10, 28, 20, 0, 0, 0, time.UTC) } + + // adopt: the export proposed whole as the seed, once. + got, err := s.doAdopt(map[string]any{"why": "take over"}, "operator@laptop") + if err != nil { + t.Fatal(err) + } + seed := got.(proposed).Proposal + if !seed.Seed || seed.ExportDigest != delivery.ExportDigest(map[string]delivery.Placement{"anchor": anchor}) || !slices.Equal(asked, []string{exportCall}) || + len(st[delivery.PlacementCandidatesState].values) != 1 || st[delivery.PlacementProposalsState].values[seed.ID] == nil { + t.Fatalf("adopt = %+v, asked %v", seed, asked) + } + if _, err := s.doAdopt(map[string]any{"why": "again"}, "op"); err == nil || !strings.Contains(err.Error(), "is open") { + t.Errorf("a second adopt with a seed open: %v", err) + } + // withdraw and again mark the proposal; the holder ends or judges it. + if _, err := s.doWithdraw(map[string]any{"id": seed.ID, "why": "not yet"}, "op"); err != nil { + t.Fatal(err) + } + if _, err := s.doAgain(map[string]any{"id": seed.ID}); err != nil { + t.Fatal(err) + } + if p, err := s.proposal(seed.ID); err != nil || !strings.Contains(p.Withdrawn, "not yet") || p.Again.IsZero() { + t.Errorf("marked = %+v, %v", p, err) + } + if _, err := s.doWithdraw(map[string]any{"id": "p-none", "why": "x"}, "op"); err == nil { + t.Error("an unknown proposal was withdrawn") + } + delete(st[delivery.PlacementProposalsState].values, seed.ID) // the holder ended it + st[delivery.PlacementCandidatesState].values = map[string][]byte{} + + // propose: refused whole, naming every finding; a valid move laid over the keys at their revisions. + raw, _ := delivery.EncodePlacement(anchor) + st[delivery.PlacementState].values["placement.anchor"] = raw + if _, err := s.doPropose(map[string]any{"why": "x", "changes": `[{"op":"place","node":"anchor","module":"ghost"},{"op":"hold","node":"anchor","module":"web","claim":"x"}]`}, "op"); err == nil || + !strings.Contains(err.Error(), "ghost") || !strings.Contains(err.Error(), "lease write") || len(st[delivery.PlacementProposalsState].values) != 0 { + t.Errorf("a refused proposal: %v", err) + } + if _, err := s.doPropose(map[string]any{"why": "x", "changes": `[{"op":"swap","node":"anchor"}]`}, "op"); err == nil { + t.Error("an unknown change was taken") + } + got, err = s.doPropose(map[string]any{"why": "move web", "changes": []any{ + map[string]any{"op": "take-off", "node": "anchor", "module": "web"}, map[string]any{"op": "place", "node": "laptop", "module": "web"}}}, "operator@laptop") + if err != nil { + t.Fatal(err) + } + move := got.(proposed).Proposal + if move.By != "operator@laptop" || move.Base["anchor"] != uint64(len(raw)) || move.Base["laptop"] != 0 || len(st[delivery.PlacementCandidatesState].values) != 2 { + t.Fatalf("propose = %+v", move) + } + + // placement and proposals read every state the seat keeps (§10); history names the stream for what is older. + reads := map[string]bool{} + pa, err := s.doPlacement(map[string]any{"module": "web", "history": "true"}) + if a := pa.(placementAnswer); err != nil || !slices.Equal(a.Nodes, []string{"anchor"}) || len(a.Changes) != 1 || !strings.Contains(a.Older, delivery.PlacementStream) { + t.Errorf("placement of web = %+v, %v", pa, err) + } else { + reads[delivery.PlacementState], reads["stream "+delivery.PlacementStream] = true, true + } + if _, err := s.doPlacement(map[string]any{"node": "laptop"}); err == nil { + t.Error("a node with no key was answered") + } + ps, err := s.doProposals(map[string]any{"id": move.ID}) + if list := ps.(map[string]any)["proposals"].([]openProposal); err != nil || len(list) != 1 || len(list[0].Candidates) != 2 { + t.Errorf("proposals = %+v, %v", ps, err) + } else { + reads[delivery.PlacementProposalsState], reads[delivery.PlacementCandidatesState] = true, true + } + raw0, err := readManifest() + if err != nil { + t.Fatal(err) + } + for _, state := range raw0.State { + if slices.Contains([]string{delivery.PlacementState, delivery.PlacementProposalsState, delivery.PlacementCandidatesState}, state.Name) && !reads[state.Name] { + t.Errorf("no verb reads the seat's state %s", state.Name) + } + } + for _, stream := range raw0.Streams { + if !reads["stream "+stream.Name] { + t.Errorf("no verb reads the seat's stream %s", stream.Name) + } + } + + // 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) + } + 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) + } +} + +// readManifest is the module's manifest, read as the controller reads it. +func readManifest() (catalogue.Manifest, error) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + return catalogue.Manifest{}, err + } + return catalogue.ParseManifest(raw) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 4b23517..0e47bb4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -351,7 +351,7 @@ func serveTools() error { time.Sleep(wait) } }() - return stdio.Serve("", s.tools()) + return stdio.Serve("", append(s.tools(), newSeatTools(runtimeStore{}, stdio.Ask).tools()...)) } // shortFP is a fingerprint's first twelve hex digits. diff --git a/modules/mesh-delivery-assigner/internal/assign/seat.go b/modules/mesh-delivery-assigner/internal/assign/seat.go new file mode 100644 index 0000000..1a7e247 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/seat.go @@ -0,0 +1,281 @@ +package assign + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The seat mesh-placement (novox/hq ADR 0567), held by the assigner beside its wants: placement is its state +// `placement`, changed only by a proposal (§2) the nodes it touches answer (§3), judged (§4) and written on the +// operator's Approve at the revision it was laid over (§5). What here decides; the tools and the passes do. + +// Leased says whether the lease write of ADR 0547 has happened, after which the controller's assignments are read no +// more (§8). Not before L3: until then the holder keeps the controller in step and refuses what its verbs cannot take. +var Leased = false + +// ProposeIn is what a proposal is laid over: its id, who and why, its changes; every node's current placement and the +// revision of its key; which modules the indexer holds and which nodes have a record; and, for the seed, the export's +// digest. +type ProposeIn struct { + ID, By, Why string + At time.Time + Changes []delivery.PlacementProposalChange + Current map[string]delivery.Placement + Revisions map[string]uint64 + Held func(module string) bool + Recorded func(node string) bool + Seed bool + ExportDigest string +} + +// Propose lays a proposal over the current keys (§2): the proposal (state answering, each changed node's base revision), +// the laid placement of every node whose key it changes, and every finding, which refuses it whole: a change the +// controller's verbs cannot take before the lease write (§8), a change that does not fit its node, and what +// CheckPlacements refuses of the result laid over every placement (two modules on one host port, a provider not placed, +// a seat held once held twice, two nodes with one address, two hubs, a module the indexer does not hold, a node with no +// record). A key whose value the proposal leaves as it is changes nothing and is not laid. +func Propose(in ProposeIn) (delivery.PlacementProposal, map[string]*delivery.Placement, []string) { + p := delivery.PlacementProposal{ID: in.ID, By: strings.TrimSpace(in.By), Why: strings.TrimSpace(in.Why), At: in.At.UTC(), + Changes: in.Changes, Seed: in.Seed, ExportDigest: in.ExportDigest, Base: map[string]uint64{}, State: delivery.ProposalAnswering} + var findings []string + if p.By == "" { + findings = append(findings, "the runtime names no caller: a proposal is made by someone") + } + if p.Why == "" { + findings = append(findings, "a proposal says why: it is shown to the operator") + } + if len(in.Changes) == 0 { + findings = append(findings, "a proposal changes something: it names no change") + } + for i, c := range in.Changes { + if (c.Op == delivery.OpSeed) != in.Seed { + findings = append(findings, fmt.Sprintf("change %d: the seed is adopt's alone, and never mixed with other changes", i+1)) + continue + } + if !Leased { + if why := c.BeforeLease(); why != "" { + findings = append(findings, fmt.Sprintf("change %d: %s", i+1, why)) + } + } + } + laid, lay := delivery.LayChanges(in.Current, in.Changes) + findings = append(findings, lay...) + all := map[string]delivery.Placement{} + for node, pl := range in.Current { + all[node] = pl + } + out := map[string]*delivery.Placement{} + for node, pl := range laid { + if before, ok := in.Current[node]; ok && delivery.SamePlacement(before, pl) { + continue + } + pl := pl + all[node] = pl + out[node] = &pl + p.Base[node] = in.Revisions[node] + } + if len(findings) == 0 && len(out) == 0 { + findings = append(findings, "the changes leave every placement as it is") + } + held, recorded := in.Held, in.Recorded + if held == nil { + held = func(string) bool { return false } + } + if recorded == nil { + recorded = func(string) bool { return false } + } + for _, r := range delivery.CheckPlacements(all, held, recorded) { + findings = append(findings, r) + } + return p, out, findings +} + +// Write is one key a warrant writes (§5): the node, its key, the revision it is written at (0: it has none, so it is +// created) and its value (nil: it is deleted). +type Write struct { + Node, Key string + Base uint64 + Value *delivery.Placement +} + +// Writes is every key a proposal writes, by node order. +func Writes(p delivery.PlacementProposal, laid map[string]*delivery.Placement) ([]Write, error) { + var out []Write + for _, node := range p.Nodes() { + key, err := delivery.PlacementKey(node) + if err != nil { + return nil, err + } + out = append(out, Write{Node: node, Key: key, Base: p.Base[node], Value: laid[node]}) + } + return out, nil +} + +// ControllerCall is one call of the controller's seat verbs the holder makes after writing an approved proposal, until +// the lease write (§8): assign, unassign, pin or unpin, with its arguments. +type ControllerCall struct { + Verb string + Args map[string]any + // Node and Module name the pair the call keeps in step (a pin names no module). + Node, Module string +} + +// Key names the call, as the condition of its pair says it. +func (c ControllerCall) Key() string { + return fmt.Sprintf("mesh-controller.%s %s %s", c.Verb, c.Node, c.Module) +} + +// ControllerCalls is every call that brings the controller's assignments and pins to the placement after: assign for a +// module placed on a node before not, unassign for one taken off, pin for a pin set or changed, unpin for one removed, +// each naming the proposal and its warrant in its why. In node order, unassigns before assigns, so a module moved +// between nodes is first taken off. +func ControllerCalls(before map[string]delivery.Placement, after map[string]*delivery.Placement, proposal, warrant string) []ControllerCall { + why := fmt.Sprintf("placement proposal %s, approved (%s)", proposal, warrant) + nodes := make([]string, 0, len(after)) + for node := range after { + nodes = append(nodes, node) + } + sort.Strings(nodes) + var off, on, pins []ControllerCall + for _, node := range nodes { + was, now := before[node], delivery.Placement{} + if a := after[node]; a != nil { + now = *a + } + had, has := map[string]bool{}, map[string]bool{} + for _, m := range was.Modules { + had[m.Module] = true + } + for _, m := range now.Modules { + has[m.Module] = true + if !had[m.Module] { + on = append(on, ControllerCall{Verb: "assign", Node: node, Module: m.Module, Args: map[string]any{"node": node, "module": m.Module}}) + } + } + for _, m := range was.Modules { + if !has[m.Module] { + off = append(off, ControllerCall{Verb: "unassign", Node: node, Module: m.Module, Args: map[string]any{"node": node, "module": m.Module}}) + } + } + for _, provision := range sortedKeys(now.Pins) { + to := now.Pins[provision] + if old, ok := was.Pins[provision]; !ok || old != to { + pins = append(pins, ControllerCall{Verb: "pin", Node: node, Module: provision, + Args: map[string]any{"node": node, "provision": provision, "from": to.Node, "module": to.Module}}) + } + } + for _, provision := range sortedKeys(was.Pins) { + if _, ok := now.Pins[provision]; !ok { + pins = append(pins, ControllerCall{Verb: "unpin", Node: node, Module: provision, Args: map[string]any{"node": node, "provision": provision}}) + } + } + } + out := append(append(off, on...), pins...) + for i := range out { + out[i].Args["why"] = why + } + return out +} + +// PlacementAct is the act an approved proposal's yes binds: writing exactly its changes at its base revisions. +func PlacementAct(p delivery.PlacementProposal) asks.Act { + return asks.Act{"placement": "write", "proposal": p.ID, "changes": delivery.ChangesDigest(p)} +} + +// PlacementAsking is a proposal's asking number: each judging that raises its ask again (again, a key moved) is +// another asking, under an id of its own. +func PlacementAsking(p delivery.PlacementProposal, at time.Time) int { + return int(at.Unix()) +} + +// 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 delivery.PlacementProposal, brief string, asking int, now time.Time) (delivery.AssignerAsk, error) { + binds, err := asks.ActDigest(PlacementAct(p)) + if err != nil { + return delivery.AssignerAsk{}, err + } + nodes := p.Nodes() + headline := fmt.Sprintf("Change placement on %s?", strings.Join(nodes, ", ")) + if p.Seed { + headline = "Take placement over from the controller?" + } + if n := len([]rune(headline)); n > asks.HeadlineLength { + headline = fmt.Sprintf("Change placement on %d machines?", len(nodes)) + } + return delivery.AssignerAsk{Ask: asks.Ask{ + ID: delivery.AskingID(key, asking), + Headline: headline, + Explanation: brief, + Options: []asks.Option{ + {ID: OptionYes, Label: "Approve", Does: "the placement of " + strings.Join(nodes, ", ") + " is written as proposed", + Level: asks.Approve, Binds: binds}, + {ID: OptionNo, Label: "Decline", Does: "nothing is written, and the proposal ends", Level: asks.Acknowledge}, + }, + Who: asks.Operator, + Expires: now.Add(asks.ApproveLasts), + OnExpiry: "nothing is written, and the proposal ends", + About: key, + }, Asking: asking}, nil +} + +// ExportOf reads the controller's answer to placement-export (§7): its verb answers what the command printed, the +// JSON of every node's placement, in `output` (the controller's verbAnswer; `answer` when it carries it as data). A +// refused or unreadable answer is an error naming it. +func ExportOf(raw []byte) (map[string]delivery.Placement, error) { + var a struct { + Output string `json:"output"` + OK *bool `json:"ok"` + Answer json.RawMessage `json:"answer"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("the controller's placement-export answered no verb answer: %w", err) + } + 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) + } + out := map[string]delivery.Placement{} + for node, rawP := range byNode { + p, err := delivery.DecodePlacement(rawP) + if err != nil { + return nil, fmt.Errorf("the controller's placement-export of %s cannot be read: %w", node, err) + } + if p.Node != node { + return nil, fmt.Errorf("the controller's placement-export names %s under %s", p.Node, node) + } + out[node] = p + } + if len(out) == 0 { + return nil, fmt.Errorf("the controller's placement-export names no node") + } + return out, nil +} + +// SeedChanges is the seed's changes: every node's placement of the export, whole, in node order. +func SeedChanges(export map[string]delivery.Placement) []delivery.PlacementProposalChange { + var out []delivery.PlacementProposalChange + for _, node := range sortedKeys(export) { + p := export[node] + out = append(out, delivery.PlacementProposalChange{Op: delivery.OpSeed, Node: node, Whole: &p}) + } + return out +} diff --git a/modules/mesh-delivery-assigner/internal/assign/seat_test.go b/modules/mesh-delivery-assigner/internal/assign/seat_test.go new file mode 100644 index 0000000..accec8f --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/seat_test.go @@ -0,0 +1,108 @@ +package assign + +import ( + "encoding/json" + "slices" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ADR 0567 §2, §5, §7, §8: what the seat's holder decides of a proposal: every refusal named at once, a valid one laid +// with the base revision of each key it changes, the writes at those revisions, the controller's calls that keep its +// assignments in step, the authorising ask bound to the changes, and the export read as the controller answers it. +func TestThePlacementSeatDecides(t *testing.T) { + anchor := delivery.Placement{V: delivery.Version, Node: "anchor", Decided: delivery.NodeDecided{Hub: true, Address: "10.0.0.1"}, + Modules: []delivery.PlacedModule{{Module: "store", Ports: map[string]int{"5432": 5432}}, {Module: "web"}}} + laptop := delivery.Placement{V: delivery.Version, Node: "laptop", Decided: delivery.NodeDecided{Address: "10.0.0.2"}, Modules: []delivery.PlacedModule{{Module: "shell"}}} + in := func(changes ...delivery.PlacementProposalChange) ProposeIn { + return ProposeIn{ID: "p-one", By: "operator@laptop", Why: "move web", At: time.Unix(1, 0), Changes: changes, + Current: map[string]delivery.Placement{"anchor": anchor, "laptop": laptop}, Revisions: map[string]uint64{"anchor": 4, "laptop": 9}, + Held: func(m string) bool { return m != "ghost" }, + Recorded: func(n string) bool { return n == "anchor" || n == "laptop" }} + } + move := []delivery.PlacementProposalChange{{Op: delivery.OpTakeOff, Node: "anchor", Module: "web"}, {Op: delivery.OpPlace, Node: "laptop", Module: "web"}} + p, laid, findings := Propose(in(move...)) + if len(findings) != 0 || p.State != delivery.ProposalAnswering || p.Base["anchor"] != 4 || p.Base["laptop"] != 9 || + !slices.Equal(laid["laptop"].Assigned(), []string{"shell", "web"}) || !slices.Equal(laid["anchor"].Assigned(), []string{"store"}) { + t.Fatalf("a move: %+v, %+v, %v", p, laid, findings) + } + if _, err := delivery.EncodePlacementProposal(p); err != nil { + t.Fatalf("the proposal cannot be written: %v", err) + } + for name, c := range map[string]struct { + in ProposeIn + want string + }{ + "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 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"}, + "no why": {func() ProposeIn { i := in(move...); i.Why = " "; return i }(), "says why"}, + "no caller": {func() ProposeIn { i := in(move...); i.By = ""; return i }(), "names no caller"}, + "a seed by hand": {in(delivery.PlacementProposalChange{Op: delivery.OpSeed, Node: "laptop", Whole: &laptop}), "adopt's alone"}, + "a pin of nobody placed": {in(delivery.PlacementProposalChange{Op: delivery.OpPin, Node: "laptop", Provision: "db", Provider: &delivery.Provider{Node: "laptop", Module: "store"}}), "not placed there"}, + } { + if _, _, f := Propose(c.in); !slices.ContainsFunc(f, func(s string) bool { return strings.Contains(s, c.want) }) { + t.Errorf("%s: findings %v, want one naming %q", name, f, c.want) + } + } + + 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) + } + pinned := *laid["laptop"] + pinned.Pins = map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}} + calls := ControllerCalls(map[string]delivery.Placement{"anchor": anchor, "laptop": laptop}, + map[string]*delivery.Placement{"anchor": laid["anchor"], "laptop": &pinned}, "p-one", "w1@t") + var said []string + for _, c := range calls { + said = append(said, c.Key()) + if !strings.Contains(c.Args["why"].(string), "p-one") { + t.Errorf("%s does not name the proposal: %v", c.Key(), c.Args) + } + } + if !slices.Equal(said, []string{"mesh-controller.unassign anchor web", "mesh-controller.assign laptop web", "mesh-controller.pin laptop db"}) { + t.Errorf("the controller's calls = %v", said) + } + + a, err := PlacementAsk("placement.p-one", p, "move web (proposed by operator@laptop)", 7, time.Unix(100, 0)) + if err != nil || a.Ask.Check(time.Unix(100, 0)) != nil || a.Ask.Options[0].Level != asks.Approve || a.Ask.Options[0].Performs(PlacementAct(p)) != nil { + t.Fatalf("the ask = %+v, %v", a, err) + } + q := p + q.Changes = append(slices.Clone(p.Changes), delivery.PlacementProposalChange{Op: delivery.OpUnpin, Node: "laptop", Provision: "x"}) + if a.Ask.Options[0].Performs(PlacementAct(q)) == nil { + t.Error("the ask's yes binds changes it was not asked about") + } + + export, _ := json.Marshal(map[string]any{"anchor": anchor, "laptop": laptop}) + for name, c := range map[string]struct { + raw string + want int + }{ + "in output": {`{"ok":true,"output":` + quote(string(export)+"\n") + `}`, 2}, + "as answer": {`{"ok":true,"output":"","answer":` + string(export) + `}`, 2}, + "refused": {`{"ok":false,"output":"no store"}`, 0}, + "no node": {`{"ok":true,"output":"{}"}`, 0}, + "no version": {`{"ok":true,"output":"{\"anchor\":{\"node\":\"anchor\",\"modules\":[]}}"}`, 0}, + } { + got, err := ExportOf([]byte(c.raw)) + if len(got) != c.want || (c.want == 0) != (err != nil) { + t.Errorf("%s: %d nodes, %v", name, len(got), err) + } + } + if seed := SeedChanges(map[string]delivery.Placement{"laptop": laptop, "anchor": anchor}); len(seed) != 2 || seed[0].Node != "anchor" || seed[0].Op != delivery.OpSeed { + t.Errorf("the seed's changes = %+v", seed) + } +} + +func quote(s string) string { + raw, _ := json.Marshal(s) + return string(raw) +}