novox/mesh-controller
17
0
Fork 0
Code Issues Pull Requests 3 Releases Wiki Activity

stage 7 of 8 for issue 1389

Browse Source
This commit is contained in:
mesh-tools
2026-10-20 11:17:08 +00:00
parent f77c130dcd
commit 382bc25b18
9 changed files with 66 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/seat_pass.go
+21 -69
View File
@@ -4,7 +4,6 @@ import (
"context"
"fmt"
"io"
"os"
"github.com/nats-io/nats.go"
@@ -13,24 +12,12 @@ import (
"github.com/novox/mesh-controller/internal/link"
)
// seatPassing is what passing seats on needs of the outside: the record's compare-and-set, and the hand act.
type seatPassing struct {
pass func(ctx context.Context, p catalogue.SeatPass) (bool, error)
act func(ctx context.Context, p catalogue.SeatPass) error
say io.Writer
}
// passSeatsOn records, in the composition that finds it, the new holder of each seat held once whose holder on
// record is assigned nowhere and which has exactly one claimant able to hold it, and writes the controller's own
// hand act for it (novox/hq ADR 0449 §4, issue 1317). It answers the holdings as they are after it, so the
// resolution that follows reads the record it just moved. A failed write is said and leaves the record as it was:
// a composition is never refused for it.
func passSeatsOn(ctx context.Context, sp seatPassing, holdings []catalogue.Held,
assignments []catalogue.Assignment) []catalogue.Held {
passes, said := catalogue.SeatsToPassOn(holdings, assignments)
for _, line := range said {
fmt.Fprintln(sp.say, line)
}
// performSeatPasses records, in the send that finds them, the new holder of each seat held once whose holder on record is
// assigned nowhere and which has exactly one claimant able to hold it (the composition names them: catalogue.SeatsToPassOn,
// novox/hq ADR 0449 §4, issue 1317), and writes the controller's own hand act for each. A failed write is said and leaves the
// record as it was: a send is never refused for it. It answers whether it recorded any.
func performSeatPasses(ctx context.Context, sp seatPassing, passes []catalogue.SeatPass) bool {
did := false
for _, p := range passes {
moved, err := sp.pass(ctx, p)
if err != nil {
@@ -41,21 +28,20 @@ func passSeatsOn(ctx context.Context, sp seatPassing, holdings []catalogue.Held,
if !moved {
continue // another composition passed it first, and wrote the act
}
did = true
if err := sp.act(ctx, p); err != nil {
fmt.Fprintf(sp.say, "%s passed from %s on %s to %s on %s, and the hand act could NOT be recorded: %v\n",
p.Seat.Name, p.From.Module, p.From.Node, p.To.Module, p.To.Node, err)
}
var next []catalogue.Held
for _, h := range holdings {
if hs, ok := catalogue.SeatNamed(h.Claim); ok && hs.Name == p.Seat.Name {
continue
}
next = append(next, h)
}
p.To.Site = p.From.Site
holdings = append(next, p.To)
}
return holdings
return did
}
// seatPassing is what passing seats on needs of the outside: the record's compare-and-set, and the hand act.
type seatPassing struct {
pass func(ctx context.Context, p catalogue.SeatPass) (bool, error)
act func(ctx context.Context, p catalogue.SeatPass) error
say io.Writer
}
// seatPassHandAct is the controller's own entry for a pass: it is the act that puts the holder on record,
@@ -73,31 +59,10 @@ func seatPassHandAct(p catalogue.SeatPass) link.HandAct {
}
}
// assignmentsOf are the assignments of every machine, as the passing of a seat reads them. **A module assigned and
// missing from the shelf is still assigned**: it is an assignment with no claims, so its holder is not taken for gone.
func assignmentsOf(shelf map[string]catalogue.Manifest, assigned map[string][]string) []catalogue.Assignment {
var out []catalogue.Assignment
for node, modules := range assigned {
for _, module := range modules {
m, ok := shelf[module]
if !ok {
m = catalogue.Manifest{Module: module}
}
out = append(out, catalogue.Assignment{Node: node, Manifest: m})
}
}
return out
}
// seatPassingOn is how a composition on this inventory passes a seat: against the store's record with the hand act
// on the bus, or, on a throwaway mesh (the merge gate's), in memory only, writing nothing anywhere.
// seatPassingOn is how a send passes a seat: against the store's record, with the hand act on the bus; on a throwaway mesh
// (the merge gate's) the record is kept in its store and no act is written, for there is no bus to write it to.
func seatPassingOn(inv *inventory.Inventory, w io.Writer) seatPassing {
if inv.Throwaway() {
return seatPassing{say: w,
pass: func(context.Context, catalogue.SeatPass) (bool, error) { return true, nil },
act: func(context.Context, catalogue.SeatPass) error { return nil }}
}
return seatPassing{
sp := seatPassing{
say: w,
pass: func(ctx context.Context, p catalogue.SeatPass) (bool, error) {
return inv.PassSeat(ctx, p.Seat.Name, p.Seat.Scope, p.From, p.To)
@@ -109,21 +74,8 @@ func seatPassingOn(inv *inventory.Inventory, w io.Writer) seatPassing {
})
},
}
}
// theMeshPassesSeatsOn is passSeatsOn as the composition runs it. **Only when every machine's assignments were
// read** (assignedErr nil): a machine whose read failed would look as if its modules were assigned nowhere, and
// its holder would lose the seat. Said to where the composition says its other lines (sayLeftOut).
func theMeshPassesSeatsOn(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest,
assigned map[string][]string, assignedErr error, holdings []catalogue.Held) []catalogue.Held {
var w io.Writer = os.Stderr
if lw, ok := ctx.Value(leftOutKey{}).(io.Writer); ok {
w = lw
if inv.Throwaway() {
sp.act = func(context.Context, catalogue.SeatPass) error { return nil }
}
if assignedErr != nil {
fmt.Fprintf(w, "no seat is passed on in this composition: a machine's assignments could not be read, so a "+
"holder on record might look assigned nowhere: %v\n", assignedErr)
return holdings
}
return passSeatsOn(ctx, seatPassingOn(inv, w), holdings, assignmentsOf(shelf, assigned))
return sp
}
cmd/mesh-controller/seat_pass_test.go
+35 -91
View File
@@ -26,9 +26,16 @@ func withBusSeat(t *testing.T) {
catalogue.UseSeats([]catalogue.Seat{{Name: "mesh-broker", Scope: catalogue.ScopeMesh, Delivers: "mesh-bus", Decision: "test"}})
}
// 1317/AC1: the composition records the claimant as the holder and writes one hand act by the controller naming
// the seat, the former holder and the new one; the holdings it answers name the new holder.
func TestACompositionPassesTheSeatAndWritesOneHandActByTheController(t *testing.T) {
// passesFor is what the composition names to pass on for these holdings and assignments (mesh-sdk's SeatsToPassOn), with the
// lines it says.
func passesFor(holdings []catalogue.Held, as []catalogue.Assignment) ([]catalogue.SeatPass, []string) {
return catalogue.SeatsToPassOn(holdings, as)
}
// 1317/AC1: the send records the claimant as the holder and writes one hand act by the controller naming the seat, the former
// holder and the new one. (Which seats to pass, and the holdings after it, are mesh-sdk's: go/compose rest_test.go
// TestASeatWhoseHolderIsAssignedNowhereGoesToItsOnlyClaimant and TestAHolderOnRecordSettlesTheSeatInTheRestOfTheMesh.)
func TestASendPassesTheSeatAndWritesOneHandActByTheController(t *testing.T) {
withBusSeat(t)
var passed []catalogue.SeatPass
var acts []link.HandAct
@@ -44,9 +51,9 @@ func TestACompositionPassesTheSeatAndWritesOneHandActByTheController(t *testing.
},
}
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
got := passSeatsOn(t.Context(), sp, holdings, []catalogue.Assignment{{Node: "novox", Manifest: aBrokerClaiming("gate")}})
if len(got) != 1 || got[0].Module != "gate" || got[0].Node != "novox" {
t.Fatalf("the holdings after the composition do not name the gate: %+v", got)
passes, _ := passesFor(holdings, []catalogue.Assignment{{Node: "novox", Manifest: aBrokerClaiming("gate")}})
if !performSeatPasses(t.Context(), sp, passes) {
t.Fatal("a pass was recorded and the send does not say it made something")
}
if len(passed) != 1 || len(acts) != 1 {
t.Fatalf("passes=%d acts=%d, want one each", len(passed), len(acts))
@@ -60,67 +67,35 @@ func TestACompositionPassesTheSeatAndWritesOneHandActByTheController(t *testing.
}
}
// 1317/AC2: two claimants, nothing recorded, the refusal said.
func TestACompositionWithTwoClaimantsRecordsNoneAndSaysSo(t *testing.T) {
withBusSeat(t)
sp := seatPassing{say: &bytes.Buffer{},
pass: func(context.Context, catalogue.SeatPass) (bool, error) {
t.Fatal("a seat with two claimants was passed")
return false, nil
},
act: func(context.Context, catalogue.SeatPass) error { t.Fatal("an act was written"); return nil },
}
var said bytes.Buffer
sp.say = &said
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
got := passSeatsOn(t.Context(), sp, holdings, []catalogue.Assignment{
{Node: "novox", Manifest: aBrokerClaiming("gate")}, {Node: "ace", Manifest: aBrokerClaiming("other")}})
if len(got) != 1 || got[0].Module != "old-broker" {
t.Fatalf("the record changed: %+v", got)
}
if !strings.Contains(said.String(), "gate") || !strings.Contains(said.String(), "other") ||
!strings.Contains(said.String(), "seat mesh-broker --to") {
t.Fatalf("the refusal is not said with both and the remedy: %q", said.String())
}
}
// 1317/AC3 and AC4: nothing changes while the holder is assigned, nor with no claimant.
func TestACompositionChangesNothingWhileTheHolderIsAssignedOrNobodyClaims(t *testing.T) {
withBusSeat(t)
sp := seatPassing{say: &bytes.Buffer{},
pass: func(context.Context, catalogue.SeatPass) (bool, error) { t.Fatal("passed"); return false, nil },
act: func(context.Context, catalogue.SeatPass) error { t.Fatal("act"); return nil },
}
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
got := passSeatsOn(t.Context(), sp, holdings, []catalogue.Assignment{
{Node: "novox", Manifest: aBrokerClaiming("old-broker")}, {Node: "novox", Manifest: aBrokerClaiming("gate")}})
if len(got) != 1 || got[0].Module != "old-broker" {
t.Fatalf("holder assigned: record changed: %+v", got)
}
got = passSeatsOn(t.Context(), sp, holdings, nil)
if len(got) != 1 || got[0].Module != "old-broker" {
t.Fatalf("no claimant: record changed: %+v", got)
}
}
// A composition that lost the compare-and-set writes no act, and a failed write leaves the record.
// A send that lost the compare-and-set writes no act, and a failed write leaves the record.
func TestALostPassWritesNoActAndAFailedOneLeavesTheRecord(t *testing.T) {
withBusSeat(t)
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
as := []catalogue.Assignment{{Node: "novox", Manifest: aBrokerClaiming("gate")}}
passes, _ := passesFor(holdings, as)
lost := seatPassing{say: &bytes.Buffer{},
pass: func(context.Context, catalogue.SeatPass) (bool, error) { return false, nil },
act: func(context.Context, catalogue.SeatPass) error { t.Fatal("act without a pass"); return nil },
}
passSeatsOn(t.Context(), lost, holdings, as)
if performSeatPasses(t.Context(), lost, passes) {
t.Fatal("a lost pass is said to have made something")
}
var said bytes.Buffer
sp := seatPassing{say: &said,
pass: func(context.Context, catalogue.SeatPass) (bool, error) { return false, context.DeadlineExceeded },
act: func(context.Context, catalogue.SeatPass) error { t.Fatal("act"); return nil },
}
got := passSeatsOn(t.Context(), sp, holdings, as)
if got[0].Module != "old-broker" || !strings.Contains(said.String(), "could not be passed") {
t.Fatalf("got %+v said %q", got, said.String())
if performSeatPasses(t.Context(), sp, passes) || !strings.Contains(said.String(), "could not be passed") {
t.Fatalf("a failed pass is not said: %q", said.String())
}
// And an act that cannot be written is said, with the record kept.
var told bytes.Buffer
kept := seatPassing{say: &told,
pass: func(context.Context, catalogue.SeatPass) (bool, error) { return true, nil },
act: func(context.Context, catalogue.SeatPass) error { return errors.New("no bus") },
}
if !performSeatPasses(t.Context(), kept, passes) || !strings.Contains(told.String(), "the hand act could NOT be recorded") {
t.Fatalf("a failed act is not said: %q", told.String())
}
}
@@ -146,46 +121,15 @@ func TestTwoSeatPassesAreNoRepairAndRaiseNoHealerWanted(t *testing.T) {
}
}
// A failed read of one machine's assignments passes no seat: its holder would look assigned nowhere.
func TestNoSeatIsPassedWhenAMachinesAssignmentsCouldNotBeRead(t *testing.T) {
withBusSeat(t)
var said bytes.Buffer
ctx := leftOutSaidBy(t.Context(), &said)
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "ace", Module: "old-broker"}}
shelf := map[string]catalogue.Manifest{"gate": aBrokerClaiming("gate")}
got := theMeshPassesSeatsOn(ctx, nil, shelf, map[string][]string{"novox": {"gate"}},
errors.New("ace: the bus did not answer"), holdings)
if len(got) != 1 || got[0].Module != "old-broker" {
t.Fatalf("the record moved: %+v", got)
}
if !strings.Contains(said.String(), "no seat is passed") || !strings.Contains(said.String(), "ace") {
t.Errorf("not said: %q", said.String())
}
}
// A holder that is assigned but missing from the shelf is still assigned.
func TestAHolderAssignedButNotOnTheShelfIsNotTakenForGone(t *testing.T) {
withBusSeat(t)
as := assignmentsOf(map[string]catalogue.Manifest{"gate": aBrokerClaiming("gate")},
map[string][]string{"novox": {"gate", "old-broker"}})
passes, _ := catalogue.SeatsToPassOn(
[]catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}, as)
if len(passes) != 0 {
t.Fatalf("a holder missing from the shelf lost its seat: %+v", passes)
}
}
// The merge gate's throwaway mesh passes the seat in memory and writes nothing: no record, no hand act.
func TestAThrowawayMeshPassesTheSeatInMemoryAndWritesNothing(t *testing.T) {
withBusSeat(t)
// On the merge gate's throwaway mesh the pass is recorded in its store and no hand act is written: there is no bus to write
// it to.
func TestAThrowawayMeshWritesNoHandActForAPass(t *testing.T) {
inv := &inventory.Inventory{}
inv.MarkThrowaway()
var said bytes.Buffer
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
got := passSeatsOn(t.Context(), seatPassingOn(inv, &said), holdings,
[]catalogue.Assignment{{Node: "novox", Manifest: aBrokerClaiming("gate")}})
if len(got) != 1 || got[0].Module != "gate" {
t.Fatalf("the gate's view does not hold the pass: %+v", got)
sp := seatPassingOn(inv, &said)
if err := sp.act(t.Context(), catalogue.SeatPass{}); err != nil {
t.Fatalf("the throwaway mesh tried to write the act: %v", err)
}
if said.Len() != 0 {
t.Errorf("the throwaway composition said %q (a write was tried)", said.String())
cmd/mesh-controller/sendable.go
-53
View File
@@ -1,11 +1,8 @@
package main
import (
"context"
"encoding/json"
"fmt"
"sort"
"strings"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
@@ -168,53 +165,3 @@ func (s sendable) Body() ([]byte, error) {
}
return json.Marshal(envelope)
}
// adoptionOf is the envelope for a node, nil when it is converged.
func adoptionOf(ctx context.Context, inv *inventory.Inventory, record inventory.Node,
plan catalogue.Resolution, composed catalogue.Composed) (*adoptionEnvelope, error) {
if !record.Adopted {
return nil, nil
}
taken, err := inv.Taken(ctx, record.Name)
if err != nil {
return nil, err
}
return adoptionFor(plan, taken, composed), nil
}
// adoptionFor is the envelope computed from what was taken and who owns each resource.
//
// Every module the node runs that is not taken is untaken — including one pulled in by another
// rather than assigned: what is found is kept until its module is taken, whoever put it there.
func adoptionFor(plan catalogue.Resolution, taken []string,
composed catalogue.Composed) *adoptionEnvelope {
isTaken := map[string]bool{}
for _, m := range taken {
isTaken[m] = true
}
out := &adoptionEnvelope{Taken: []string{}}
runs := map[string]bool{}
for _, m := range plan.Modules {
runs[m.Module] = true
if isTaken[m.Module] {
out.Taken = append(out.Taken, m.Module)
}
}
sort.Strings(out.Taken)
for _, r := range composed.Resources {
id, _ := r["id"].(string)
module, owned := composed.Owner[id]
// Every kind, not only files and containers (novox/hq ADR 0103): a directory, a service, a
// container mounting what was found and an action run in a held container all reach what
// the machine already has. What the mesh declares of its own is never held.
if !owned || !runs[module] || isTaken[module] ||
strings.HasPrefix(id, catalogue.AdoptionPrefix) {
continue
}
if out.Untaken == nil {
out.Untaken = map[string][]string{}
}
out.Untaken[module] = append(out.Untaken[module], id)
}
return out
}
cmd/mesh-controller/sendable_integration_test.go
+3 -3
View File
@@ -157,7 +157,7 @@ func TestConsumersAreToldTheGivenPort(t *testing.T) {
t.Fatal(err)
}
plan, _, err := planFor(ctx, open, "laptop")
plan, _, err := resolveNode(ctx, open, "laptop")
if err != nil {
t.Fatal(err)
}
@@ -183,7 +183,7 @@ func TestConsumersAreToldTheGivenPort(t *testing.T) {
!strings.Contains(err.Error(), "per node") {
t.Fatalf("a port given for the whole mesh was not refused: %v", err)
}
plan, settings, err := planFor(ctx, open, "anchor")
plan, settings, err := resolveNode(ctx, open, "anchor")
if err != nil {
t.Fatal(err)
}
@@ -283,7 +283,7 @@ func TestTheMachineSideOfAMappingIsMovedEverywhereTheNumberIsUsed(t *testing.T)
}
// And the consumer on the other machine dials the same number.
plan, _, err := planFor(ctx, open, "laptop")
plan, _, err := resolveNode(ctx, open, "laptop")
if err != nil {
t.Fatal(err)
}
cmd/mesh-controller/sendable_test.go
+1 -1
View File
@@ -26,7 +26,7 @@ func helloWeb() catalogue.Manifest {
// composed is what node would be sent now, as push composes it.
func composed(t *testing.T, open *stores, node string) sendable {
t.Helper()
plan, settings, err := planFor(t.Context(), open, node)
plan, settings, err := resolveNode(t.Context(), open, node)
if err != nil {
t.Fatal(err)
}
cmd/mesh-controller/sent_required_by_integration_test.go
+1 -1
View File
@@ -38,7 +38,7 @@ func TestAModulePulledInIsHandedOnAMachineThatCanTakeItAndMirrored(t *testing.T)
t.Fatal(err)
}
plan, settings, err := planFor(ctx, open, "laptop")
plan, settings, err := resolveNode(ctx, open, "laptop")
if err != nil {
t.Fatal(err)
}
cmd/mesh-controller/served_own_names_test.go
+2 -2
View File
@@ -36,7 +36,7 @@ func TestAProviderElsewhereServesItsOwnComposedName(t *testing.T) {
if err := open.inventory.SetPublicDomain(ctx, "anchor", "example.test"); err != nil {
t.Fatal(err)
}
plan, _, err := planFor(ctx, open, "laptop")
plan, _, err := resolveNode(ctx, open, "laptop")
if err != nil {
t.Fatal(err)
}
@@ -48,7 +48,7 @@ func TestAProviderElsewhereServesItsOwnComposedName(t *testing.T) {
if err := open.inventory.SetPublicDomain(ctx, "anchor", ""); err != nil {
t.Fatal(err)
}
plan, _, err = planFor(ctx, open, "laptop")
plan, _, err = resolveNode(ctx, open, "laptop")
if err != nil {
t.Fatal(err)
}
cmd/mesh-controller/unbound_test.go
+2 -2
View File
@@ -24,7 +24,7 @@ func TestPlanAndNodeListAnOptionalRequirementLeftUnbound(t *testing.T) {
t.Fatalf("grafana could not be assigned on a mesh without influxdb: %v", err)
}
plan, _, err := planFor(ctx, open, "laptop")
plan, _, err := resolveNode(ctx, open, "laptop")
if err != nil {
t.Fatal(err)
}
@@ -44,7 +44,7 @@ func TestPlanAndNodeListAnOptionalRequirementLeftUnbound(t *testing.T) {
if _, err := assign(ctx, open, "anchor", "influxdb"); err != nil {
t.Fatal(err)
}
plan, _, err = planFor(ctx, open, "laptop")
plan, _, err = resolveNode(ctx, open, "laptop")
if err != nil {
t.Fatal(err)
}
cmd/mesh-controller/vault_switch_test.go
+1 -1
View File
@@ -271,7 +271,7 @@ func TestAnIssuedValueTakesThePathOfItsProvidersModule(t *testing.T) {
}
}
// The consumer's send makes the pair, so the provision has a holder to take it for.
plan, settings, err := planFor(ctx, open, "laptop")
plan, settings, err := resolveNode(ctx, open, "laptop")
if err != nil {
t.Fatal(err)
}