Merge pull request 'Placement keyed per node and module, no ASSIGNMENTS stream, and plan entries carrying a consumer's contribution (issue 774)' (#148) from feat/072-moves into main
This commit was merged in pull request #148.
This commit is contained in:
+17
-12
@@ -187,12 +187,17 @@ func HeldCredential(w WantedAccount, password string) ([]byte, error) {
|
|||||||
Claims: w.Claims})
|
Claims: w.Claims})
|
||||||
}
|
}
|
||||||
|
|
||||||
// **One issuer at every moment: minting turns on at the lease write** (novox/hq ADR 0561 §3, §6). The controller's lease
|
// **The lease write** (novox/hq ADR 0561 §3, §6). The controller's lease is the one key `holder` of its bucket
|
||||||
// is the one key `holder` of its bucket (LeaseBucket). The operator's credential hands it away by writing a value that
|
// (LeaseBucket). The operator's credential hands it away by writing a value that names its new holder in `handed-to`;
|
||||||
// names its new holder in `handed-to`; while the key holds that, the vault mints and the grants part creates the durable
|
// while the key holds that, the grants part creates the durable consumers an account needs, the assigner reads the grants
|
||||||
// consumers an account needs. A controller's value, a key deleted or expired, or one that cannot be read or parsed, turn
|
// part's grants (delivery.GrantSource), and the vault retires and rotates accounts. A controller's value, a key deleted or
|
||||||
// both off: when in doubt nothing is minted, so the controller and the vault never both mint one account. No clock and
|
// expired, or one that cannot be read or parsed, turn those off. No clock and no setting turns it.
|
||||||
// no setting turns it.
|
//
|
||||||
|
// **Minting is no longer on the lease** (issue 774, research 072's gap "vault mints after the lease"): the vault is the
|
||||||
|
// one issuer of every account the grants part wants, whoever holds the lease, and the controller mints none once a vault
|
||||||
|
// is installed (it forwards `module issue` to the vault's issue-account). One issuer at every moment holds by who mints,
|
||||||
|
// not by the lease: an account whose value the vault holds is never minted anew, and a value the controller hands or
|
||||||
|
// moves never replaces one the vault minted.
|
||||||
|
|
||||||
// ControllerLease is the lease's bucket as a reader names it; the vault and the grants part alone read it
|
// ControllerLease is the lease's bucket as a reader names it; the vault and the grants part alone read it
|
||||||
// (NamedReaders), the controller's announcement marking it readable by nobody.
|
// (NamedReaders), the controller's announcement marking it readable by nobody.
|
||||||
@@ -221,24 +226,24 @@ type LeaseHolder struct {
|
|||||||
func LeaseHandedAway(raw []byte, found bool, readErr error) (bool, string) {
|
func LeaseHandedAway(raw []byte, found bool, readErr error) (bool, string) {
|
||||||
switch {
|
switch {
|
||||||
case readErr != nil:
|
case readErr != nil:
|
||||||
return false, "the controller's lease cannot be read (" + readErr.Error() + "), so nothing is minted"
|
return false, "the controller's lease cannot be read (" + readErr.Error() + "), so it is not handed away"
|
||||||
case !found || len(raw) == 0 || string(raw) == "null":
|
case !found || len(raw) == 0 || string(raw) == "null":
|
||||||
return false, "no lease is written (expired, deleted, or never taken): a controller may take it, so nothing is minted"
|
return false, "no lease is written (expired, deleted, or never taken): a controller may take it, so it is not handed away"
|
||||||
}
|
}
|
||||||
var h LeaseHolder
|
var h LeaseHolder
|
||||||
if err := json.Unmarshal(raw, &h); err != nil {
|
if err := json.Unmarshal(raw, &h); err != nil {
|
||||||
return false, "the controller's lease is not readable as a holder (" + err.Error() + "), so nothing is minted"
|
return false, "the controller's lease is not readable as a holder (" + err.Error() + "), so it is not handed away"
|
||||||
}
|
}
|
||||||
if strings.TrimSpace(h.HandedTo) == "" {
|
if strings.TrimSpace(h.HandedTo) == "" {
|
||||||
who := h.Instance
|
who := h.Instance
|
||||||
if who == "" {
|
if who == "" {
|
||||||
who = "an instance that names itself not"
|
who = "an instance that names itself not"
|
||||||
}
|
}
|
||||||
return false, "the controller holds the lease (" + who + "): it mints, so the vault mints nothing (ADR 0561 §3)"
|
return false, "the controller holds the lease (" + who + "), so it is not handed away (ADR 0561 §3)"
|
||||||
}
|
}
|
||||||
if h.Instance != "" {
|
if h.Instance != "" {
|
||||||
return false, "the lease names both a controller instance (" + h.Instance + ") and a holder it is handed to (" +
|
return false, "the lease names both a controller instance (" + h.Instance + ") and a holder it is handed to (" +
|
||||||
h.HandedTo + "), so nothing is minted"
|
h.HandedTo + "), so it is not handed away"
|
||||||
}
|
}
|
||||||
return true, "the lease is handed to " + h.HandedTo + ": the vault mints every new account (ADR 0561 §1, §3)"
|
return true, "the lease is handed to " + h.HandedTo + " (ADR 0561 §3)"
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ import (
|
|||||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||||
"github.com/nats-io/nats.go/jetstream"
|
"github.com/nats-io/nats.go/jetstream"
|
||||||
)
|
)
|
||||||
@@ -67,6 +68,9 @@ func mayBeyondTheControllers(user string) map[string]bool {
|
|||||||
key = parts[0] + ".*"
|
key = parts[0] + ".*"
|
||||||
}
|
}
|
||||||
out := map[string]bool{}
|
out := map[string]bool{}
|
||||||
|
if membership.GrantsBucket != GrantsMembershipsBucket || membership.GrantsKey("a", "b") != MembershipKey("a", "b") {
|
||||||
|
panic("the membership package's grants bucket does not agree with buscompose's") // FollowMembership reads it
|
||||||
|
}
|
||||||
for _, s := range readGrants(GrantsMembershipsBucket, key) {
|
for _, s := range readGrants(GrantsMembershipsBucket, key) {
|
||||||
out[s] = true
|
out[s] = true
|
||||||
}
|
}
|
||||||
@@ -151,11 +155,16 @@ func grantsAsTheControllers(t *testing.T, got, want string) {
|
|||||||
}
|
}
|
||||||
intended := beyondTheControllers(user)
|
intended := beyondTheControllers(user)
|
||||||
for g := range was {
|
for g := range was {
|
||||||
if !now[g] {
|
// A runtime's direct get of its membership from the assignments stream went with the stream (research 072
|
||||||
|
// row 33): it reads the grants part's state memberships instead (ADR 0571 clause 7).
|
||||||
|
if !now[g] && !strings.HasPrefix(g, "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment."+strings.SplitN(user, ".", 2)[0]+".") {
|
||||||
t.Errorf("%s: the copy does not grant %s, which the controller's does", user, g)
|
t.Errorf("%s: the copy does not grant %s, which the controller's does", user, g)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for g := range now {
|
for g := range now {
|
||||||
|
if strings.HasPrefix(g, "$JS.API.DIRECT.GET."+AssignmentsStream+".") && !strings.HasSuffix(user, "."+GrantsModule) {
|
||||||
|
t.Errorf("%s: the copy grants %s, a read of the assignments stream, which goes at the lease write", user, g)
|
||||||
|
}
|
||||||
if !was[g] && !intended[g] && !mayBeyondTheControllers(user)[g] && !createdAdvisory.MatchString(`"`+g+`", `) &&
|
if !was[g] && !intended[g] && !mayBeyondTheControllers(user)[g] && !createdAdvisory.MatchString(`"`+g+`", `) &&
|
||||||
!createsAnAccountsConsumer(user, g) && !engineBeyond(user, g) {
|
!createsAnAccountsConsumer(user, g) && !engineBeyond(user, g) {
|
||||||
t.Errorf("%s: the copy grants %s beyond the controller's, and no decision names it", user, g)
|
t.Errorf("%s: the copy grants %s beyond the controller's, and no decision names it", user, g)
|
||||||
@@ -686,7 +695,7 @@ func healthChecks(t *testing.T) {
|
|||||||
{Module: "telegram", Holds: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}},
|
{Module: "telegram", Holds: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}},
|
||||||
Processes: []ProcessPrincipal{{Process: "w", Queues: []QueueOwner{{Process: "w", Stream: "M_WORK"}, {Process: "w", Stream: "KV_x"}}}}},
|
Processes: []ProcessPrincipal{{Process: "w", Queues: []QueueOwner{{Process: "w", Stream: "M_WORK"}, {Process: "w", Stream: "KV_x"}}}}},
|
||||||
{Module: "nats", SnapshotsTheBus: true, Processes: []ProcessPrincipal{{Process: "nats-tools",
|
{Module: "nats", SnapshotsTheBus: true, Processes: []ProcessPrincipal{{Process: "nats-tools",
|
||||||
Removes: []string{catalogue.RemovesControllerDeadLetterConsumer, catalogue.RemovesEngineActs}}, {Process: "other"}}},
|
Removes: []string{catalogue.RemovesControllerDeadLetterConsumer, catalogue.RemovesEngineActs, catalogue.RemovesAssignmentsStream}}, {Process: "other"}}},
|
||||||
{Module: "shop", Processes: []ProcessPrincipal{{Process: "nats-tools", Removes: []string{catalogue.RemovesEngineActs}}}},
|
{Module: "shop", Processes: []ProcessPrincipal{{Process: "nats-tools", Removes: []string{catalogue.RemovesEngineActs}}}},
|
||||||
}}})
|
}}})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -722,7 +731,9 @@ func healthChecks(t *testing.T) {
|
|||||||
WatchmanModule + " sources": {"$JS.API.STREAM.MSG.GET.EVENTS", "$JS.API.STREAM.INFO.SEAT_TELEGRAM_SENDER",
|
WatchmanModule + " sources": {"$JS.API.STREAM.MSG.GET.EVENTS", "$JS.API.STREAM.INFO.SEAT_TELEGRAM_SENDER",
|
||||||
"$JS.API.STREAM.MSG.GET.M_WORK", "$JS.API.STREAM.INFO.EVENTS", "mesh.events.dead.EVENTS.a_m"},
|
"$JS.API.STREAM.MSG.GET.M_WORK", "$JS.API.STREAM.INFO.EVENTS", "mesh.events.dead.EVENTS.a_m"},
|
||||||
"a.nats.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller",
|
"a.nats.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller",
|
||||||
"$KV.mesh-host_acts.g14.a1", "$JS.API.STREAM.INFO.KV_mesh-host_acts"},
|
"$KV.mesh-host_acts.g14.a1", "$JS.API.STREAM.INFO.KV_mesh-host_acts",
|
||||||
|
// Research 072 row 33: the ASSIGNMENTS stream is read and removed at the lease write.
|
||||||
|
"$JS.API.STREAM.INFO.ASSIGNMENTS", "$JS.API.STREAM.DELETE.ASSIGNMENTS"},
|
||||||
}
|
}
|
||||||
mayNot := map[string][]string{
|
mayNot := map[string][]string{
|
||||||
// The keeper writes no findings and no attempts; no observer writes the keeper's buckets (clause 2).
|
// The keeper writes no findings and no attempts; no observer writes the keeper's buckets (clause 2).
|
||||||
@@ -763,10 +774,11 @@ func healthChecks(t *testing.T) {
|
|||||||
// remove nothing; an account is granted the removals its bus declares and no other (the catalogue refuses them
|
// remove nothing; an account is granted the removals its bus declares and no other (the catalogue refuses them
|
||||||
// outside the bus module).
|
// outside the bus module).
|
||||||
"a.nats.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.other", "$JS.API.CONSUMER.DELETE.EVENTS.a_m",
|
"a.nats.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.other", "$JS.API.CONSUMER.DELETE.EVENTS.a_m",
|
||||||
"$KV.mesh-conditions_conditions.x", "$JS.API.STREAM.DELETE.KV_mesh-host_acts", "$JS.API.STREAM.PURGE.KV_mesh-host_acts"},
|
"$KV.mesh-conditions_conditions.x", "$JS.API.STREAM.DELETE.KV_mesh-host_acts", "$JS.API.STREAM.PURGE.KV_mesh-host_acts",
|
||||||
|
"$JS.API.STREAM.DELETE.EVENTS", "$JS.API.STREAM.PURGE.ASSIGNMENTS", "$JS.API.STREAM.DELETE.ASSIGNMENTS_X"},
|
||||||
"a.nats.other": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-host_acts.g14.a1"},
|
"a.nats.other": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-host_acts.g14.a1"},
|
||||||
"a.nats": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-host_acts.g14.a1"},
|
"a.nats": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-host_acts.g14.a1"},
|
||||||
"a.shop.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-conditions_conditions.x"},
|
"a.shop.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-conditions_conditions.x", "$JS.API.STREAM.DELETE.ASSIGNMENTS"},
|
||||||
}
|
}
|
||||||
// ADR 0571 clause 1: a read naming another module's log is issued to the reader as a log it reads, never writes.
|
// ADR 0571 clause 1: a read naming another module's log is issued to the reader as a log it reads, never writes.
|
||||||
if m := MembershipFor("a", Declared{Module: delivery.IndexerModule, Reads: []string{delivery.BuildLogRead, "x.y"}},
|
if m := MembershipFor("a", Declared{Module: delivery.IndexerModule, Reads: []string{delivery.BuildLogRead, "x.y"}},
|
||||||
|
|||||||
@@ -71,6 +71,8 @@ func TestTheGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
|
|||||||
{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_undeclared", false},
|
{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_undeclared", false},
|
||||||
{GrantsModule, "$JS.API.STREAM.CREATE.EVENTS", false},
|
{GrantsModule, "$JS.API.STREAM.CREATE.EVENTS", false},
|
||||||
{GrantsModule, "$JS.API.STREAM.CREATE.ASSIGNMENTS", false},
|
{GrantsModule, "$JS.API.STREAM.CREATE.ASSIGNMENTS", false},
|
||||||
|
// Research 072 row 33: a runtime reads its membership from the grants part's state, never the assignments stream.
|
||||||
|
{"mesh-delivery-builder", "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.novox.mesh-delivery-builder", false},
|
||||||
{GrantsModule, "$JS.API.STREAM.INFO.EVENTS", false},
|
{GrantsModule, "$JS.API.STREAM.INFO.EVENTS", false},
|
||||||
{"mesh-delivery-builder", "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_steps", false},
|
{"mesh-delivery-builder", "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_steps", false},
|
||||||
// The node-engine's acts: created by the grants part, written by each node's engine under its own keys alone.
|
// The node-engine's acts: created by the grants part, written by each node's engine under its own keys alone.
|
||||||
|
|||||||
@@ -40,7 +40,8 @@ var (
|
|||||||
// BusRemovalGrants are the whole authority of an account whose bus declares removals (ADR 0564 Limits 4, issue 1952
|
// BusRemovalGrants are the whole authority of an account whose bus declares removals (ADR 0564 Limits 4, issue 1952
|
||||||
// notes 6 and 7), each removal its grant: the controller's consumer on DEAD_LETTER_NOTICES read and deleted (the lease
|
// notes 6 and 7), each removal its grant: the controller's consumer on DEAD_LETTER_NOTICES read and deleted (the lease
|
||||||
// write's step, clause 5); the node-engine's acts bucket read (STREAM.INFO with a subject filter lists what a removal
|
// write's step, clause 5); the node-engine's acts bucket read (STREAM.INFO with a subject filter lists what a removal
|
||||||
// would take, so the tools read before they remove) and its keys removed, which nothing else may empty when it fills. A
|
// would take, so the tools read before they remove) and its keys removed, which nothing else may empty when it fills; the ASSIGNMENTS stream read and deleted whole (the lease
|
||||||
|
// write, research 072 row 33). A
|
||||||
// removal outside the set is refused, as the catalogue refuses it (catalogue.BusRemovals).
|
// removal outside the set is refused, as the catalogue refuses it (catalogue.BusRemovals).
|
||||||
func BusRemovalGrants(removes []string) ([]string, error) {
|
func BusRemovalGrants(removes []string) ([]string, error) {
|
||||||
var out []string
|
var out []string
|
||||||
@@ -51,6 +52,8 @@ func BusRemovalGrants(removes []string) ([]string, error) {
|
|||||||
"$JS.API.CONSUMER.DELETE."+DeadLetterNoticesStream+"."+ControllerName)
|
"$JS.API.CONSUMER.DELETE."+DeadLetterNoticesStream+"."+ControllerName)
|
||||||
case catalogue.RemovesEngineActs:
|
case catalogue.RemovesEngineActs:
|
||||||
out = append(out, "$JS.API.STREAM.INFO.KV_"+EngineActsBucket, "$KV."+EngineActsBucket+".>")
|
out = append(out, "$JS.API.STREAM.INFO.KV_"+EngineActsBucket, "$KV."+EngineActsBucket+".>")
|
||||||
|
case catalogue.RemovesAssignmentsStream:
|
||||||
|
out = append(out, "$JS.API.STREAM.INFO."+AssignmentsStream, "$JS.API.STREAM.DELETE."+AssignmentsStream)
|
||||||
default:
|
default:
|
||||||
return nil, fmt.Errorf("a bus account removes %q, which is not one of %s (novox/hq ADR 0564 Limits 4)", r,
|
return nil, fmt.Errorf("a bus account removes %q, which is not one of %s (novox/hq ADR 0564 Limits 4)", r,
|
||||||
strings.Join(catalogue.BusRemovals, ", "))
|
strings.Join(catalogue.BusRemovals, ", "))
|
||||||
@@ -59,6 +62,13 @@ func BusRemovalGrants(removes []string) ([]string, error) {
|
|||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// removesAssignmentsStream is an account declared to remove the stream ASSIGNMENTS: the writers table's one exemption from
|
||||||
|
// stream definitions being the controller's, for deleting that one stream at the lease write (research 072 row 33).
|
||||||
|
func removesAssignmentsStream(p Principal, pattern string) bool {
|
||||||
|
return p.Kind == KindProcess && slices.Contains(p.Removes, catalogue.RemovesAssignmentsStream) &&
|
||||||
|
pattern == "$JS.API.STREAM.DELETE."+AssignmentsStream
|
||||||
|
}
|
||||||
|
|
||||||
// removesEngineActs is an account declared to remove keys of the node-engine's acts bucket: the writers table's one
|
// removesEngineActs is an account declared to remove keys of the node-engine's acts bucket: the writers table's one
|
||||||
// exemption from a node's engine being the only writer of its keys, for removal (ADR 0564 Limits 4).
|
// exemption from a node's engine being the only writer of its keys, for removal (ADR 0564 Limits 4).
|
||||||
func removesEngineActs(p Principal, pattern string) bool {
|
func removesEngineActs(p Principal, pattern string) bool {
|
||||||
|
|||||||
@@ -1061,12 +1061,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
sub = append(sub, announcing(announced...)...)
|
sub = append(sub, announcing(announced...)...)
|
||||||
// Its own membership (ADR 0160): the one subject a runtime derives for itself, read
|
// Its own membership (ADR 0160): the one subject a runtime derives for itself, heard live. Nothing else's.
|
||||||
// directly from the stream and followed live. Nothing else's.
|
|
||||||
sub = append(sub, MembershipSubject(p.Node, p.Module))
|
sub = append(sub, MembershipSubject(p.Node, p.Module))
|
||||||
pub = append(pub, "$JS.API.DIRECT.GET."+AssignmentsStream+"."+MembershipSubject(p.Node, p.Module))
|
// And from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7): its one read of it,
|
||||||
// And from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7): the read that replaces
|
// since the assignments stream's direct get went with the stream (research 072 row 33).
|
||||||
// the stream's at the lease write, when the grants part stops publishing there.
|
|
||||||
pub = append(pub, readGrants(GrantsMembershipsBucket, MembershipKey(p.Node, p.Module))...)
|
pub = append(pub, readGrants(GrantsMembershipsBucket, MembershipKey(p.Node, p.Module))...)
|
||||||
|
|
||||||
// 1b. The tools it calls, if its manifest says it calls any (novox/hq ADR 0152). The same
|
// 1b. The tools it calls, if its manifest says it calls any (novox/hq ADR 0152). The same
|
||||||
@@ -1228,11 +1226,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Every assigned module's membership on this node (ADR 0160): one per module, read
|
// Every assigned module's membership on this node (ADR 0160): one per module, heard live and read from the
|
||||||
// directly from the stream and followed live. This node's and no other's — the one token
|
// grants part's state memberships (ADR 0571 clause 7). This node's and no other's — the one token
|
||||||
// that varies is the module, so the pattern is the machine's own assignments.
|
// that varies is the module, so the pattern is the machine's own assignments.
|
||||||
sub = append(sub, "mesh.assignment."+p.Node+".*")
|
sub = append(sub, "mesh.assignment."+p.Node+".*")
|
||||||
pub = append(pub, "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment."+p.Node+".*")
|
|
||||||
pub = append(pub, readGrants(GrantsMembershipsBucket, p.Node+".*")...)
|
pub = append(pub, readGrants(GrantsMembershipsBucket, p.Node+".*")...)
|
||||||
// And that it is there (novox/hq to-be 45 §3, S11): its own heartbeat, under its machine's
|
// And that it is there (novox/hq to-be 45 §3, S11): its own heartbeat, under its machine's
|
||||||
// name and no other's, on core NATS like the host's.
|
// name and no other's, on core NATS like the host's.
|
||||||
|
|||||||
@@ -171,14 +171,9 @@ func MeshStreams() []Stream {
|
|||||||
Why: "one declaration per node, always the newest; a node that sees sequence n refuses " +
|
Why: "one declaration per node, always the newest; a node that sees sequence n refuses " +
|
||||||
"n-1 by construction (issue 107)",
|
"n-1 by construction (issue 107)",
|
||||||
},
|
},
|
||||||
{
|
// The assignments stream (ADR 0160) is no longer the mesh's: a runtime reads its membership from the grants
|
||||||
Name: AssignmentsStream,
|
// part's state memberships (ADR 0571 clause 7), and the bus module's tools delete the stream at the lease write
|
||||||
Subjects: []string{"mesh.assignment.*.*"},
|
// (research 072 row 33), so it is not declared here, which would make it again.
|
||||||
Retention: RetentionLastPerSubject,
|
|
||||||
Direct: true,
|
|
||||||
Why: "one membership per assignment, always the newest: what the mesh issued this module " +
|
|
||||||
"on this machine to serve and to reach (ADR 0160); read directly by the runtime it is for",
|
|
||||||
},
|
|
||||||
{
|
{
|
||||||
Name: EventsStream,
|
Name: EventsStream,
|
||||||
// A seat's own events ride here too: they are 1:many like any event, and the
|
// A seat's own events ride here too: they are 1:many like any event, and the
|
||||||
|
|||||||
@@ -43,6 +43,9 @@ func createsDeclared(p Principal, pattern string) bool {
|
|||||||
if isController(p, pattern) {
|
if isController(p, pattern) {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
if removesAssignmentsStream(p, pattern) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM.CREATE.")
|
name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM.CREATE.")
|
||||||
if ok && p.Kind == KindModule && p.Module == GrantsModule && name == GrantsUserListObjectStream {
|
if ok && p.Kind == KindModule && p.Module == GrantsModule && name == GrantsUserListObjectStream {
|
||||||
return true
|
return true
|
||||||
@@ -185,7 +188,7 @@ var WritersTable = []WriterRow{
|
|||||||
Subjects: []string{"$JS.API.STREAM.CREATE.>", "$JS.API.STREAM.UPDATE.>", "$JS.API.STREAM.DELETE.>",
|
Subjects: []string{"$JS.API.STREAM.CREATE.>", "$JS.API.STREAM.UPDATE.>", "$JS.API.STREAM.DELETE.>",
|
||||||
"$JS.API.CONSUMER.DURABLE.CREATE.>"},
|
"$JS.API.CONSUMER.DURABLE.CREATE.>"},
|
||||||
Writes: createsDeclared,
|
Writes: createsDeclared,
|
||||||
Shared: "the grants part creates a stream a manifest declares, by its name, and never updates or deletes one (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54)"},
|
Shared: "the grants part creates a stream a manifest declares, by its name, and never updates or deletes one (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54); the bus module's tools account deletes the stream ASSIGNMENTS at the lease write, and no other (research 072 row 33)"},
|
||||||
{State: "builds and their outcomes", Writer: "the build seat's holder", KeptIn: "its own state",
|
{State: "builds and their outcomes", Writer: "the build seat's holder", KeptIn: "its own state",
|
||||||
Others: "the controller asks",
|
Others: "the controller asks",
|
||||||
Subjects: []string{"mesh.seat.node-build-agent.event.built", "mesh.seat.mesh-build-machine.event.built"},
|
Subjects: []string{"mesh.seat.node-build-agent.event.built", "mesh.seat.mesh-build-machine.event.built"},
|
||||||
|
|||||||
@@ -277,6 +277,10 @@ type Rendering struct {
|
|||||||
// templated files are handed to the client (handed.go); without either, it composes as before.
|
// templated files are handed to the client (handed.go); without either, it composes as before.
|
||||||
Hands bool
|
Hands bool
|
||||||
Provisioner string
|
Provisioner string
|
||||||
|
// Assigned is, by module, whether the vault judged it handed (AssignmentHands of the plan in its assignment), as the
|
||||||
|
// vault answered the node composing itself; where it names a module, Handing follows it alone (issue 774). Nil, or a
|
||||||
|
// module it does not name, is judged by the manifest and the machine, as the composer of the plans does.
|
||||||
|
Assigned map[string]bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// RootField is a user resource's field saying the account must never become root without a person
|
// RootField is a user resource's field saying the account must never become root without a person
|
||||||
@@ -1076,6 +1080,11 @@ func (r Resolution) compose(with Rendering, owner map[string]string,
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
handed.name(m.Module, to, own)
|
handed.name(m.Module, to, own)
|
||||||
|
contributed, err := r.contributedOf(m, to, with.Settings[m.Module])
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
handed.contribute(m.Module, to, contributed)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// A handed collector is filled from items alone: each old form its collections would have held is said
|
// A handed collector is filled from items alone: each old form its collections would have held is said
|
||||||
@@ -1823,6 +1832,47 @@ func (r Resolution) contributions(settings SettingsBy, grants []Grant,
|
|||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// contributedOf is each contribution m makes to one requirement, composed as its provider receives it in a received
|
||||||
|
// file (contributions: the one of `contributes` with what the module shares of its bucket, then each of several by
|
||||||
|
// local name), with where m's machine is, encoded for PlanProvision.Contributed (issue 774): one composition, so a
|
||||||
|
// plan and a file cannot say different things. Nil where m contributes nothing to it.
|
||||||
|
func (r Resolution) contributedOf(m Manifest, to string, layers []Layer) (json.RawMessage, error) {
|
||||||
|
form := ContributedForm{At: r.At}
|
||||||
|
add := func(local string, values map[string]any) error {
|
||||||
|
raw, err := json.Marshal(values)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("%s contributing to %s: %w", m.Module, to, err)
|
||||||
|
}
|
||||||
|
form.Each = append(form.Each, ContributedOne{Local: local, Values: raw})
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if raw, ok := m.Contributes[to]; ok {
|
||||||
|
values, err := r.composed(m, to, raw, layers, m.Module+" contributing to "+to)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if values, err = withShares(m, to, values); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := add("", values); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, local := range sortedKeys(m.ContributesMany[to]) {
|
||||||
|
values, err := r.composed(m, to, m.ContributesMany[to][local], layers, m.Module+" contributing "+local+" to "+to)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := add(local, values); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(form.Each) == 0 {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
return EncodeContributed(form)
|
||||||
|
}
|
||||||
|
|
||||||
// composed is one contribution as its provider receives it: settled with this node's settings, its
|
// composed is one contribution as its provider receives it: settled with this node's settings, its
|
||||||
// endpoint's port filled in, and its names composed from the label.
|
// endpoint's port filled in, and its names composed from the label.
|
||||||
//
|
//
|
||||||
|
|||||||
+130
-3
@@ -1,6 +1,7 @@
|
|||||||
package catalogue
|
package catalogue
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"maps"
|
"maps"
|
||||||
@@ -134,6 +135,80 @@ type PlanProvision struct {
|
|||||||
// alone knows the node's domains, and carried unchanged by the provider's helper into the consumer's entry. Nil
|
// alone knows the node's domains, and carried unchanged by the provider's helper into the consumer's entry. Nil
|
||||||
// where the consumer contributes nothing to the provision, and the entry is then as it was.
|
// where the consumer contributes nothing to the provision, and the entry is then as it was.
|
||||||
Names map[string]string `json:"names,omitempty"`
|
Names map[string]string `json:"names,omitempty"`
|
||||||
|
// Contributed is what the consumer gives its provider through this provision, opaque here (issue 774, research
|
||||||
|
// 072 row 34): the bytes of a ContributedForm, each contribution composed as its provider receives it and where
|
||||||
|
// the consumer's machine is. What a contribution's values mean is the provider's own contract, published by the
|
||||||
|
// provider's repository (route-proxy's for a route), which decodes them; the seat, its plans, its client and the
|
||||||
|
// helper carry them unread. On the provision's first entry alone; nil where the consumer contributes nothing to it.
|
||||||
|
Contributed json.RawMessage `json:"contributed,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ContributedVersion is the version of ContributedForm this SDK writes and reads.
|
||||||
|
const ContributedVersion = 1
|
||||||
|
|
||||||
|
// ContributedForm is the generic form of PlanProvision.Contributed: every contribution one consumer makes to one
|
||||||
|
// requirement, composed (settled with its node's settings, its endpoint's port filled in, its names joined to the
|
||||||
|
// node's domains: Resolution.composed, the one composition a received file was written from), and At, where the
|
||||||
|
// consumer's machine is on the private network, empty when it is on none. The provisioner knows no field of any
|
||||||
|
// contribution: Values are the bytes of an object in the provider's contract.
|
||||||
|
type ContributedForm struct {
|
||||||
|
V int `json:"v"`
|
||||||
|
At string `json:"at,omitempty"`
|
||||||
|
Each []ContributedOne `json:"each"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ContributedOne is one contribution: Local its local name among several (`contributes` naming several), empty for a
|
||||||
|
// requirement's one, and Values the contribution's object.
|
||||||
|
type ContributedOne struct {
|
||||||
|
Local string `json:"local,omitempty"`
|
||||||
|
Values json.RawMessage `json:"values"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// EncodeContributed is f as the bytes PlanProvision.Contributed carries, refused where DecodeContributed would refuse
|
||||||
|
// them, so a plan never carries what its reader cannot take.
|
||||||
|
func EncodeContributed(f ContributedForm) (json.RawMessage, error) {
|
||||||
|
f.V = ContributedVersion
|
||||||
|
raw, err := json.Marshal(f)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if _, err := DecodeContributed(raw); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return raw, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// DecodeContributed reads PlanProvision.Contributed strictly: this version, no field it does not know, at least one
|
||||||
|
// contribution, each local name once, each value a JSON object. What a value says is not judged here.
|
||||||
|
func DecodeContributed(raw json.RawMessage) (ContributedForm, error) {
|
||||||
|
var f ContributedForm
|
||||||
|
dec := json.NewDecoder(strings.NewReader(string(raw)))
|
||||||
|
dec.DisallowUnknownFields()
|
||||||
|
if err := dec.Decode(&f); err != nil {
|
||||||
|
return ContributedForm{}, fmt.Errorf("what a consumer contributed is not readable: %w", err)
|
||||||
|
}
|
||||||
|
if dec.More() {
|
||||||
|
return ContributedForm{}, errors.New("what a consumer contributed carries more than one document")
|
||||||
|
}
|
||||||
|
if f.V != ContributedVersion {
|
||||||
|
return ContributedForm{}, fmt.Errorf("what a consumer contributed is of version %d, this SDK reads %d", f.V,
|
||||||
|
ContributedVersion)
|
||||||
|
}
|
||||||
|
if len(f.Each) == 0 {
|
||||||
|
return ContributedForm{}, errors.New("what a consumer contributed holds no contribution")
|
||||||
|
}
|
||||||
|
seen := map[string]bool{}
|
||||||
|
for _, one := range f.Each {
|
||||||
|
if seen[one.Local] {
|
||||||
|
return ContributedForm{}, fmt.Errorf("what a consumer contributed names %q twice", one.Local)
|
||||||
|
}
|
||||||
|
seen[one.Local] = true
|
||||||
|
var object map[string]json.RawMessage
|
||||||
|
if err := json.Unmarshal(one.Values, &object); err != nil || object == nil {
|
||||||
|
return ContributedForm{}, fmt.Errorf("the contribution %q is not a JSON object", one.Local)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return f, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// provisionsOf is, from a resolution's needs, each provision module requires and its provider, by provision then
|
// provisionsOf is, from a resolution's needs, each provision module requires and its provider, by provision then
|
||||||
@@ -294,16 +369,43 @@ func handedBy(m Manifest, archive Built) *Handed {
|
|||||||
return h
|
return h
|
||||||
}
|
}
|
||||||
|
|
||||||
// handing is whether this composition hands the module's templated files to the provisioner.
|
// handing is whether this composition hands the module's templated files to the provisioner: handed (Handing), with
|
||||||
|
// the templates its build kept. A module the vault judged handed whose build kept none is refused where its secrets
|
||||||
|
// are placed ("nothing would fill it"), never composed with values the vault does not answer.
|
||||||
func (with Rendering) handing(m Manifest) bool {
|
func (with Rendering) handing(m Manifest) bool {
|
||||||
return m.Moved && m.Handed != nil && with.Hands && with.Provisioner != ""
|
return with.Handing(m) && m.Handed != nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Handing is whether a machine composed with this rendering hands the module to the provisioner, its files and
|
// Handing is whether a machine composed with this rendering hands the module to the provisioner, its files and
|
||||||
// its own secrets (design 57 P6, V4): a moved module there, assigned or only pulled in by another (issue 765), on a
|
// its own secrets (design 57 P6, V4): a moved module there, assigned or only pulled in by another (issue 765), on a
|
||||||
// machine that can take it. The controller makes and seals none of such a module's secrets for that machine: its
|
// machine that can take it. The controller makes and seals none of such a module's secrets for that machine: its
|
||||||
// client asks the vault.
|
// client asks the vault.
|
||||||
func (with Rendering) Handing(m Manifest) bool { return with.handing(m) }
|
//
|
||||||
|
// **Where the vault judged the module, its word is the rule** (Rendering.Assigned, issue 774): the vault leaves a
|
||||||
|
// module's own secrets out of `own` exactly when the plan in its assignment hands it (AssignmentHands), so a node
|
||||||
|
// composing itself hands exactly those, whatever its own reading of the manifest and the machine says. On 2026-10-31
|
||||||
|
// postgres on novox was held: its plan handed its secrets, the vault answered none of them, and the node, judging by
|
||||||
|
// the manifest and the machine, composed them itself and refused for want of `reader`. Two readers, one rule.
|
||||||
|
func (with Rendering) Handing(m Manifest) bool {
|
||||||
|
if hands, judged := with.Assigned[m.Module]; judged {
|
||||||
|
return hands
|
||||||
|
}
|
||||||
|
return m.Moved && m.Handed != nil && with.Hands && with.Provisioner != ""
|
||||||
|
}
|
||||||
|
|
||||||
|
// AssignmentHands is whether a module's value in the seat mesh-provisioner's plans (the assignments mirror's shape,
|
||||||
|
// `{"provisioning": <Plan>}`) hands its module to the provisioner: Plan.Hands of the plan it carries, false with no
|
||||||
|
// plan. A value that does not read is an error, never "not handed". The one rule the vault judges `own` by and a node
|
||||||
|
// composing itself follows (Rendering.Assigned, issue 774).
|
||||||
|
func AssignmentHands(raw []byte) (bool, error) {
|
||||||
|
var a struct {
|
||||||
|
Provisioning *Plan `json:"provisioning"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(raw, &a); err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
return a.Provisioning != nil && a.Provisioning.Hands(), nil
|
||||||
|
}
|
||||||
|
|
||||||
// requiredByWords begins why a module only pulled in by another is on a machine (Resolution.Because): `required by
|
// requiredByWords begins why a module only pulled in by another is on a machine (Resolution.Because): `required by
|
||||||
// <module>`.
|
// <module>`.
|
||||||
@@ -603,6 +705,22 @@ type handedFiles struct {
|
|||||||
// keyedOnly is each module with a plan only for what Keyed keys of a consumer not handed: a provision its
|
// keyedOnly is each module with a plan only for what Keyed keys of a consumer not handed: a provision its
|
||||||
// provider's helper answers (issue 774), such as a route.
|
// provider's helper answers (issue 774), such as a route.
|
||||||
keyedOnly map[string]bool
|
keyedOnly map[string]bool
|
||||||
|
// contributed is what each handed module gives each requirement's provider, encoded (PlanProvision.Contributed).
|
||||||
|
contributed map[string]map[string]json.RawMessage
|
||||||
|
}
|
||||||
|
|
||||||
|
// contribute records what a handed module gives one requirement's provider, encoded; nothing for none.
|
||||||
|
func (h *handedFiles) contribute(module, provision string, raw json.RawMessage) {
|
||||||
|
if len(raw) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if h.contributed == nil {
|
||||||
|
h.contributed = map[string]map[string]json.RawMessage{}
|
||||||
|
}
|
||||||
|
if h.contributed[module] == nil {
|
||||||
|
h.contributed[module] = map[string]json.RawMessage{}
|
||||||
|
}
|
||||||
|
h.contributed[module][provision] = raw
|
||||||
}
|
}
|
||||||
|
|
||||||
// keyed starts the plan of a module begin started none for, when the composition keys any of its needs as a consumer
|
// keyed starts the plan of a module begin started none for, when the composition keys any of its needs as a consumer
|
||||||
@@ -830,6 +948,15 @@ func (h *handedFiles) finished(needs []Needed) map[string]Plan {
|
|||||||
provisions[i].Names = maps.Clone(h.names[module][provisions[i].Provision])
|
provisions[i].Names = maps.Clone(h.names[module][provisions[i].Provision])
|
||||||
}
|
}
|
||||||
sortProvisions(provisions)
|
sortProvisions(provisions)
|
||||||
|
// **What the consumer contributes, on the provision's first entry** (issue 774): a provider serving from the
|
||||||
|
// plans alone reads each contribution once, however many local names the requirement has.
|
||||||
|
gave := map[string]bool{}
|
||||||
|
for i := range provisions {
|
||||||
|
if raw := h.contributed[module][provisions[i].Provision]; len(raw) > 0 && !gave[provisions[i].Provision] {
|
||||||
|
gave[provisions[i].Provision] = true
|
||||||
|
provisions[i].Contributed = slices.Clone(raw)
|
||||||
|
}
|
||||||
|
}
|
||||||
plan.Provisions = provisions
|
plan.Provisions = provisions
|
||||||
h.plans[module] = plan
|
h.plans[module] = plan
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -187,7 +187,9 @@ func TestAHandedModulesPlanEntryCarriesWhatItContributesToItsProvider(t *testing
|
|||||||
want := []PlanProvision{{Provision: "postgres-database", Provider: "anchor", ProviderModule: "store",
|
want := []PlanProvision{{Provision: "postgres-database", Provider: "anchor", ProviderModule: "store",
|
||||||
CredentialForm: CredentialFormPair, Contributes: map[string]any{"name": "nextcloud"}, At: "anchor.internal", From: "anchor", As: ConsumerIdentity("workstation", "notify"),
|
CredentialForm: CredentialFormPair, Contributes: map[string]any{"name": "nextcloud"}, At: "anchor.internal", From: "anchor", As: ConsumerIdentity("workstation", "notify"),
|
||||||
// What the binding file says as `${bound:postgres-database:name}` (issue 792).
|
// What the binding file says as `${bound:postgres-database:name}` (issue 792).
|
||||||
Names: map[string]string{"name": "nextcloud"}}}
|
Names: map[string]string{"name": "nextcloud"},
|
||||||
|
// And what a received file carried of it, composed, with where its machine is: opaque bytes (issue 774).
|
||||||
|
Contributed: json.RawMessage(`{"v":1,"at":"workstation.internal","each":[{"values":{"name":"nextcloud"}}]}`)}}
|
||||||
if got := c.Plans["notify"].Provisions; !reflect.DeepEqual(got, want) {
|
if got := c.Plans["notify"].Provisions; !reflect.DeepEqual(got, want) {
|
||||||
t.Fatalf("the plan names %+v, want %+v", got, want)
|
t.Fatalf("the plan names %+v, want %+v", got, want)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -65,6 +65,15 @@ func TestAHandedModulesSecretsAreNamedAndNoValueIsSealed(t *testing.T) {
|
|||||||
if want := []PlanSecret{{Name: "token", Path: path}}; !reflect.DeepEqual(c.Plans["bot"].Secrets, want) {
|
if want := []PlanSecret{{Name: "token", Path: path}}; !reflect.DeepEqual(c.Plans["bot"].Secrets, want) {
|
||||||
t.Fatalf("the plan names secrets %+v, want %+v", c.Plans["bot"].Secrets, want)
|
t.Fatalf("the plan names secrets %+v, want %+v", c.Plans["bot"].Secrets, want)
|
||||||
}
|
}
|
||||||
|
// Issue 774 (postgres on novox, 2026-10-31): where the vault judged it handed, its secrets are the vault's even on a
|
||||||
|
// machine the node reads as not taking a filled file, and nothing is refused for a value the vault never answers.
|
||||||
|
judged, err := r.Compose(Rendering{ArtifactStore: "store:5000", Assigned: map[string]bool{"bot": true}})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("a module the vault judged handed is refused: %v", err)
|
||||||
|
}
|
||||||
|
if want := []PlanSecret{{Name: "token", Path: path}}; !reflect.DeepEqual(judged.Plans["bot"].Secrets, want) {
|
||||||
|
t.Errorf("judged handed by the vault, the plan names secrets %+v, want %+v", judged.Plans["bot"].Secrets, want)
|
||||||
|
}
|
||||||
// A process naming the token's file still restarts when its client writes it.
|
// A process naming the token's file still restarts when its client writes it.
|
||||||
for _, res := range c.Resources {
|
for _, res := range c.Resources {
|
||||||
if res["id"] == "bot.daemon" && !strings.Contains(strings.Join(stringsIn(res["restart-on"]), " "), "bot.needs-token") {
|
if res["id"] == "bot.daemon" && !strings.Contains(strings.Join(stringsIn(res["restart-on"]), " "), "bot.needs-token") {
|
||||||
@@ -99,6 +108,9 @@ func TestAMovedModulesSecretsAreSealedAsBeforeWhereItIsNotHanded(t *testing.T) {
|
|||||||
for name, with := range map[string]Rendering{
|
for name, with := range map[string]Rendering{
|
||||||
"an engine that does not take a filled file alone": {Provisioner: "mesh-provisioner-client", ArtifactStore: "store:5000"},
|
"an engine that does not take a filled file alone": {Provisioner: "mesh-provisioner-client", ArtifactStore: "store:5000"},
|
||||||
"no provisioner client on the machine": {Hands: true, ArtifactStore: "store:5000"},
|
"no provisioner client on the machine": {Hands: true, ArtifactStore: "store:5000"},
|
||||||
|
// Issue 774: where the vault judged the module not handed, the node composes it so, whatever else it reads.
|
||||||
|
"the vault judged it not handed": {Hands: true, Provisioner: "mesh-provisioner-client", ArtifactStore: "store:5000",
|
||||||
|
Assigned: map[string]bool{"bot": false}},
|
||||||
} {
|
} {
|
||||||
c, got := composeWith(t, notifierOn(t, held(t, botMoved(), builtBot(botText))), sealedForBot(with))
|
c, got := composeWith(t, notifierOn(t, held(t, botMoved(), builtBot(botText))), sealedForBot(with))
|
||||||
if got != before {
|
if got != before {
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package catalogue
|
package catalogue
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"encoding/json"
|
||||||
"reflect"
|
"reflect"
|
||||||
"slices"
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -75,8 +76,9 @@ func TestAModuleNotHandedHasAPlanForWhatAHelperAnswers(t *testing.T) {
|
|||||||
consumer := notifierConsumer(t, false, false)
|
consumer := notifierConsumer(t, false, false)
|
||||||
consumer.Requires = append(consumer.Requires, "route")
|
consumer.Requires = append(consumer.Requires, "route")
|
||||||
consumer.Contributes = map[string]map[string]any{"route": {"host": "notify.example.org"}}
|
consumer.Contributes = map[string]map[string]any{"route": {"host": "notify.example.org"}}
|
||||||
|
consumer.ContributesMany = map[string]map[string]map[string]any{"route": {"api": {"label": "notify-api", "port": 9000}}}
|
||||||
node := workstation()
|
node := workstation()
|
||||||
node.At = "workstation.internal"
|
node.At, node.PublicDomain = "workstation.internal", "example.org"
|
||||||
proxy := Manifest{Module: "proxy", Version: "1", Provides: []Offer{{Name: "route", Scope: ScopeMesh, AnsweredBy: AnsweredByHelper}}}
|
proxy := Manifest{Module: "proxy", Version: "1", Provides: []Offer{{Name: "route", Scope: ScopeMesh, AnsweredBy: AnsweredByHelper}}}
|
||||||
store := Manifest{Module: "store", Version: "1", Provides: []Offer{{Name: "postgres-database", Scope: ScopeMesh,
|
store := Manifest{Module: "store", Version: "1", Provides: []Offer{{Name: "postgres-database", Scope: ScopeMesh,
|
||||||
Credential: &OfferCredential{Form: CredentialFormPair}}}, Serves: map[string]map[string]any{"postgres-database": {"port": 5432}}}
|
Credential: &OfferCredential{Form: CredentialFormPair}}}, Serves: map[string]map[string]any{"postgres-database": {"port": 5432}}}
|
||||||
@@ -104,6 +106,29 @@ func TestAModuleNotHandedHasAPlanForWhatAHelperAnswers(t *testing.T) {
|
|||||||
!strings.Contains(plan.Provisions[0].As, "notify") || plan.Provisions[0].Contributes["host"] != "notify.example.org" {
|
!strings.Contains(plan.Provisions[0].As, "notify") || plan.Provisions[0].Contributes["host"] != "notify.example.org" {
|
||||||
t.Fatalf("the plan names %+v", plan)
|
t.Fatalf("the plan names %+v", plan)
|
||||||
}
|
}
|
||||||
|
// What it contributes, composed as a received file carried it, and where its machine is, as opaque bytes the
|
||||||
|
// provider's own contract reads (issue 774, research 072 row 34): route-proxy serves the route from the plan alone.
|
||||||
|
given, err := DecodeContributed(plan.Provisions[0].Contributed)
|
||||||
|
if err != nil || given.At != "workstation.internal" || len(given.Each) != 2 || given.Each[0].Local != "" ||
|
||||||
|
string(given.Each[0].Values) != `{"host":"notify.example.org"}` || given.Each[1].Local != "api" ||
|
||||||
|
string(given.Each[1].Values) != `{"internal-name":"notify-api.anchor.internal","label":"notify-api","name":"notify-api.example.org","port":9000}` {
|
||||||
|
t.Fatalf("the plan gives %+v (%v)", given, err)
|
||||||
|
}
|
||||||
|
// Read strictly: a form the SDK does not write is refused, whatever its contributions say.
|
||||||
|
for raw, refused := range map[string]string{
|
||||||
|
`{"v":2,"each":[{"values":{}}]}`: "version 2",
|
||||||
|
`{"v":1,"each":[]}`: "no contribution",
|
||||||
|
`{"v":1,"each":[{"values":{}}],"route":{}}`: "unknown field",
|
||||||
|
`{"v":1,"each":[{"values":{}},{"values":{}}]}`: "twice",
|
||||||
|
`{"v":1,"each":[{"local":"api","values":["a"]}]}`: "not a JSON object",
|
||||||
|
`{"v":1,"each":[{"values":{}}]} {}`: "more than one",
|
||||||
|
`{"v":1,"at":"10.0.0.2","each":[{"values":{"x":true}}]}`: "",
|
||||||
|
} {
|
||||||
|
_, err := DecodeContributed(json.RawMessage(raw))
|
||||||
|
if (refused == "") != (err == nil) || err != nil && !strings.Contains(err.Error(), refused) {
|
||||||
|
t.Errorf("DecodeContributed(%s) = %v, want %q", raw, err, refused)
|
||||||
|
}
|
||||||
|
}
|
||||||
// And it hands nothing: Plan.Hands, the one rule every reader of a plan asks, by each part a plan can name.
|
// And it hands nothing: Plan.Hands, the one rule every reader of a plan asks, by each part a plan can name.
|
||||||
if plan.Hands() {
|
if plan.Hands() {
|
||||||
t.Fatalf("a plan naming the route alone hands its module: %+v", plan)
|
t.Fatalf("a plan naming the route alone hands its module: %+v", plan)
|
||||||
|
|||||||
@@ -41,10 +41,13 @@ type ProcessBus struct {
|
|||||||
const (
|
const (
|
||||||
RemovesControllerDeadLetterConsumer = "DEAD_LETTER_NOTICES.controller"
|
RemovesControllerDeadLetterConsumer = "DEAD_LETTER_NOTICES.controller"
|
||||||
RemovesEngineActs = "mesh-host_acts"
|
RemovesEngineActs = "mesh-host_acts"
|
||||||
|
// RemovesAssignmentsStream is the stream ASSIGNMENTS, removed whole at the lease write (novox/hq research 072 row 33):
|
||||||
|
// its 350 messages have no consumer, and the placement is the provisioner's from then on.
|
||||||
|
RemovesAssignmentsStream = "ASSIGNMENTS"
|
||||||
)
|
)
|
||||||
|
|
||||||
// BusRemovals is the closed set of what a bus account may remove.
|
// BusRemovals is the closed set of what a bus account may remove.
|
||||||
var BusRemovals = []string{RemovesControllerDeadLetterConsumer, RemovesEngineActs}
|
var BusRemovals = []string{RemovesControllerDeadLetterConsumer, RemovesEngineActs, RemovesAssignmentsStream}
|
||||||
|
|
||||||
// BusAccount is one account of a module's own beside its `broker`: a process's (ADR 0436) or a tools bundle's (ADR 0564
|
// BusAccount is one account of a module's own beside its `broker`: a process's (ADR 0436) or a tools bundle's (ADR 0564
|
||||||
// Limits 4), by its id (the process resource's, or the bundle's binary), with what its bus says.
|
// Limits 4), by its id (the process resource's, or the bundle's binary), with what its bus says.
|
||||||
|
|||||||
@@ -63,7 +63,7 @@ func TestAProcessDeclaresItsBusAndIsAccepted(t *testing.T) {
|
|||||||
for _, c := range []struct {
|
for _, c := range []struct {
|
||||||
name, manifest, refused string
|
name, manifest, refused string
|
||||||
}{
|
}{
|
||||||
{name: "the bus module's tools removing", manifest: tools(broker, `{"removes":["DEAD_LETTER_NOTICES.controller","mesh-host_acts"]}`)},
|
{name: "the bus module's tools removing", manifest: tools(broker, `{"removes":["DEAD_LETTER_NOTICES.controller","mesh-host_acts","ASSIGNMENTS"]}`)},
|
||||||
{name: "a removal outside the set", manifest: tools(broker, `{"removes":["EVENTS.a_m"]}`), refused: "not one of"},
|
{name: "a removal outside the set", manifest: tools(broker, `{"removes":["EVENTS.a_m"]}`), refused: "not one of"},
|
||||||
{name: "a removal beside a grant", manifest: tools(broker, `{"removes":["mesh-host_acts"],"invokes":["nats.nats_streams"]}`), refused: "with nothing else"},
|
{name: "a removal beside a grant", manifest: tools(broker, `{"removes":["mesh-host_acts"],"invokes":["nats.nats_streams"]}`), refused: "with nothing else"},
|
||||||
{name: "another module's tools removing", manifest: tools(`[]`, `{"removes":["mesh-host_acts"]}`), refused: "only the bus module"},
|
{name: "another module's tools removing", manifest: tools(`[]`, `{"removes":["mesh-host_acts"]}`), refused: "only the bus module"},
|
||||||
@@ -88,7 +88,7 @@ func TestAProcessDeclaresItsBusAndIsAccepted(t *testing.T) {
|
|||||||
}
|
}
|
||||||
accounts := m.BusAccounts()
|
accounts := m.BusAccounts()
|
||||||
if len(accounts) != 1 || accounts[0].ID != "nats-tools" || !accounts[0].Tools ||
|
if len(accounts) != 1 || accounts[0].ID != "nats-tools" || !accounts[0].Tools ||
|
||||||
!strSliceEqual(accounts[0].Bus.Removes, []string{RemovesControllerDeadLetterConsumer, RemovesEngineActs}) {
|
!strSliceEqual(accounts[0].Bus.Removes, []string{RemovesControllerDeadLetterConsumer, RemovesEngineActs, RemovesAssignmentsStream}) {
|
||||||
t.Errorf("%s: the accounts are %+v", c.name, accounts)
|
t.Errorf("%s: the accounts are %+v", c.name, accounts)
|
||||||
}
|
}
|
||||||
// Its secret's file is the runtime account's, as the runtime's own credential is; the module's other secrets stay
|
// Its secret's file is the runtime account's, as the runtime's own credential is; the module's other secrets stay
|
||||||
|
|||||||
@@ -48,7 +48,7 @@ var ReservedStreamPrefixes = []string{"SEAT_", "LOG_", "KV_", "OBJ_", "DEAD_"}
|
|||||||
|
|
||||||
// MeshStreamNames are the mesh's own streams by name. Written here as well as in the broker, which this
|
// MeshStreamNames are the mesh's own streams by name. Written here as well as in the broker, which this
|
||||||
// package cannot import; a test keeps them agreeing.
|
// package cannot import; a test keeps them agreeing.
|
||||||
var MeshStreamNames = []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS", "PROV_APPLIED", "CHECKS_READY", "CHECKS_OTHER"}
|
var MeshStreamNames = []string{"CONTROL", "NODES", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS", "PROV_APPLIED", "CHECKS_READY", "CHECKS_OTHER"}
|
||||||
|
|
||||||
// StreamDeclaration is one stream a module owns.
|
// StreamDeclaration is one stream a module owns.
|
||||||
type StreamDeclaration struct {
|
type StreamDeclaration struct {
|
||||||
|
|||||||
@@ -94,6 +94,15 @@ type SignersReader interface {
|
|||||||
VaultAuthority(ctx context.Context) string
|
VaultAuthority(ctx context.Context) string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// HandedReader is a Reader that also says, of the composing node's modules, which the vault judged handed to the
|
||||||
|
// provisioner when it answered OwnSecrets (the vault's `own`, its `handed`: catalogue.AssignmentHands of the plan in
|
||||||
|
// each module's assignment). The node's composition then hands exactly those (compose.Node.Handed, issue 774), so the
|
||||||
|
// vault and the node never judge one module two ways. Nil, or a Reader without it (a vault older than the field),
|
||||||
|
// composes by the manifest and the machine, as before.
|
||||||
|
type HandedReader interface {
|
||||||
|
Handed(ctx context.Context, node string) map[string]bool
|
||||||
|
}
|
||||||
|
|
||||||
// Mesh is what is true of the whole mesh and is no part's state on the bus yet: the private network's range and the
|
// Mesh is what is true of the whole mesh and is no part's state on the bus yet: the private network's range and the
|
||||||
// internal names' suffix as genesis was told them, and the mesh's authority certificate (public). Empty fields take
|
// internal names' suffix as genesis was told them, and the mesh's authority certificate (public). Empty fields take
|
||||||
// compose's defaults.
|
// compose's defaults.
|
||||||
@@ -408,6 +417,16 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module
|
|||||||
if in.OwnSecrets, err = r.OwnSecrets(ctx, s.Node); err != nil {
|
if in.OwnSecrets, err = r.OwnSecrets(ctx, s.Node); err != nil {
|
||||||
return compose.Input{}, fmt.Errorf("frombus: the own secrets of %s's modules: %w", s.Node, err)
|
return compose.Input{}, fmt.Errorf("frombus: the own secrets of %s's modules: %w", s.Node, err)
|
||||||
}
|
}
|
||||||
|
// And which of its modules the vault left out of that answer as handed: the node hands exactly those.
|
||||||
|
if hr, ok := r.(HandedReader); ok {
|
||||||
|
if handed := hr.Handed(ctx, s.Node); handed != nil {
|
||||||
|
for i := range in.Nodes {
|
||||||
|
if in.Nodes[i].Name == s.Node {
|
||||||
|
in.Nodes[i].Handed = handed
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
// The core's credentials travel as the vault's box sealed to the machine (ADR 0349 §16): the same values, which
|
// The core's credentials travel as the vault's box sealed to the machine (ADR 0349 §16): the same values, which
|
||||||
// the composition reads for a core module as its kept boxes.
|
// the composition reads for a core module as its kept boxes.
|
||||||
|
|||||||
@@ -27,6 +27,7 @@ import (
|
|||||||
// bus is a Reader over values held in memory, counting whose config values were read.
|
// bus is a Reader over values held in memory, counting whose config values were read.
|
||||||
type bus struct {
|
type bus struct {
|
||||||
vaultAuthority string
|
vaultAuthority string
|
||||||
|
handed map[string]bool
|
||||||
placements map[string]delivery.Placement
|
placements map[string]delivery.Placement
|
||||||
records map[string]delivery.NodeRecord
|
records map[string]delivery.NodeRecord
|
||||||
modules map[string]delivery.Module
|
modules map[string]delivery.Module
|
||||||
@@ -79,6 +80,7 @@ func (b *bus) Placements(context.Context) (map[string]delivery.Placement, error)
|
|||||||
return b.placements, nil
|
return b.placements, nil
|
||||||
}
|
}
|
||||||
func (b *bus) VaultAuthority(context.Context) string { return b.vaultAuthority }
|
func (b *bus) VaultAuthority(context.Context) string { return b.vaultAuthority }
|
||||||
|
func (b *bus) Handed(context.Context, string) map[string]bool { return b.handed }
|
||||||
func (b *bus) Records(context.Context) (map[string]delivery.NodeRecord, error) { return b.records, nil }
|
func (b *bus) Records(context.Context) (map[string]delivery.NodeRecord, error) { return b.records, nil }
|
||||||
func (b *bus) Modules(context.Context) (map[string]delivery.Module, error) { return b.modules, nil }
|
func (b *bus) Modules(context.Context) (map[string]delivery.Module, error) { return b.modules, nil }
|
||||||
func (b *bus) Manifest(_ context.Context, digest string) ([]byte, error) {
|
func (b *bus) Manifest(_ context.Context, digest string) ([]byte, error) {
|
||||||
@@ -245,6 +247,13 @@ func TestInput(t *testing.T) {
|
|||||||
bundled.Authority != "the mesh's certificate\nthe vault's certificate\n" {
|
bundled.Authority != "the mesh's certificate\nthe vault's certificate\n" {
|
||||||
t.Errorf("the authority with the vault's published: %q, %v", bundled.Authority, err)
|
t.Errorf("the authority with the vault's published: %q, %v", bundled.Authority, err)
|
||||||
}
|
}
|
||||||
|
// Issue 774: the modules the vault judged handed when it answered own are the node's to hand, and no other node's.
|
||||||
|
judged := mk()
|
||||||
|
judged.handed = map[string]bool{"media": true, "xterm": false}
|
||||||
|
if in, err := Input(ctx, judged, Mesh{}, Scope{Node: "laptop"}); err != nil ||
|
||||||
|
!reflect.DeepEqual(in.Nodes[1].Handed, judged.handed) || in.Nodes[0].Handed != nil {
|
||||||
|
t.Errorf("the vault's handed verdicts: %+v, %v", in.Nodes, err)
|
||||||
|
}
|
||||||
if !reflect.DeepEqual(b.valuesRead, []string{"laptop." + overlay.Name, "laptop.xterm", "laptop.media"}) {
|
if !reflect.DeepEqual(b.valuesRead, []string{"laptop." + overlay.Name, "laptop.xterm", "laptop.media"}) {
|
||||||
t.Errorf("laptop read config values %v; a node reads its own slice alone", b.valuesRead)
|
t.Errorf("laptop read config values %v; a node reads its own slice alone", b.valuesRead)
|
||||||
}
|
}
|
||||||
@@ -819,11 +828,12 @@ func statesOf(t *testing.T, ctx context.Context, b *bus) mapStates {
|
|||||||
out[state][key] = raw
|
out[state][key] = raw
|
||||||
}
|
}
|
||||||
placements, _ := b.Placements(ctx)
|
placements, _ := b.Placements(ctx)
|
||||||
for node, p := range placements {
|
for _, p := range placements {
|
||||||
k, _ := delivery.PlacementKey(node)
|
values, err := delivery.PlacementValues(p)
|
||||||
raw, err := delivery.EncodePlacement(p)
|
for k, raw := range values {
|
||||||
put(PlacementRead, k, raw, err)
|
put(PlacementRead, k, raw, err)
|
||||||
}
|
}
|
||||||
|
}
|
||||||
records, _ := b.Records(ctx)
|
records, _ := b.Records(ctx)
|
||||||
for node, r := range records {
|
for node, r := range records {
|
||||||
k, _ := delivery.HaveNodeKey(node)
|
k, _ := delivery.HaveNodeKey(node)
|
||||||
|
|||||||
@@ -68,15 +68,11 @@ func (r statesReader) each(read string, visit func(key string, value []byte)) er
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (r statesReader) Placements(context.Context) (map[string]delivery.Placement, error) {
|
func (r statesReader) Placements(context.Context) (map[string]delivery.Placement, error) {
|
||||||
out := map[string]delivery.Placement{}
|
keys := delivery.NewPlacementKeys()
|
||||||
err := r.each(PlacementRead, func(k string, v []byte) {
|
err := r.each(PlacementRead, func(k string, v []byte) {
|
||||||
if node, err := delivery.ParsePlacementKey(k); err == nil {
|
_, _ = keys.Put(k, v, 0, false) // a key that does not read places nothing, as before
|
||||||
if p, err := delivery.DecodePlacement(v); err == nil && p.Node == node {
|
|
||||||
out[node] = p
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
})
|
||||||
return out, err
|
return keys.All(), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r statesReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
|
func (r statesReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
|
||||||
|
|||||||
@@ -142,6 +142,10 @@ type Node struct {
|
|||||||
Place Place
|
Place Place
|
||||||
// Assigned is the modules assigned to it, in the order the controller reads them (inv.Assigned).
|
// Assigned is the modules assigned to it, in the order the controller reads them (inv.Assigned).
|
||||||
Assigned []string
|
Assigned []string
|
||||||
|
// Handed is, by module, whether the vault judged it handed to the provisioner when it answered this node's own
|
||||||
|
// secrets (catalogue.AssignmentHands of the plan in its assignment): the node's own composition hands exactly those
|
||||||
|
// (catalogue.Rendering.Assigned, issue 774). Nil where the vault said nothing of it, as for the controller.
|
||||||
|
Handed map[string]bool
|
||||||
// Capabilities are the ones present on it (inv.ProfileOf).
|
// Capabilities are the ones present on it (inv.ProfileOf).
|
||||||
Capabilities map[string]bool
|
Capabilities map[string]bool
|
||||||
// PublicDomain is the domain it composes its routed names under (inv.PublicDomainOf).
|
// PublicDomain is the domain it composes its routed names under (inv.PublicDomainOf).
|
||||||
|
|||||||
@@ -65,7 +65,7 @@ func (c *composer) render(nodeName string, mode Mode) (Rendered, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// And each module's own secrets: a module handed to the provisioner has its secrets from the vault through the client.
|
// And each module's own secrets: a module handed to the provisioner has its secrets from the vault through the client.
|
||||||
handing := catalogue.Rendering{Hands: r.Hands, Provisioner: r.Provisioner}
|
handing := catalogue.Rendering{Hands: r.Hands, Provisioner: r.Provisioner, Assigned: r.Assigned}
|
||||||
needed, foreseenOwn, err := c.ownSecretsFor(node, plan, handing, mode)
|
needed, foreseenOwn, err := c.ownSecretsFor(node, plan, handing, mode)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Rendered{}, err
|
return Rendered{}, err
|
||||||
|
|||||||
@@ -73,6 +73,7 @@ func (c *composer) places(nodeName string, resolved Resolved) (Placed, error) {
|
|||||||
// not sent the field they would refuse.
|
// not sent the field they would refuse.
|
||||||
r.Hands = takesHanded(node)
|
r.Hands = takesHanded(node)
|
||||||
r.Provisioner = provisionerAccount(plan)
|
r.Provisioner = provisionerAccount(plan)
|
||||||
|
r.Assigned = node.Handed
|
||||||
r.ReadsHealth = node.Stated && node.Contract >= ReadinessContract
|
r.ReadsHealth = node.Stated && node.Contract >= ReadinessContract
|
||||||
r.JudgesRoot = node.Stated && node.Contract >= RootContract
|
r.JudgesRoot = node.Stated && node.Contract >= RootContract
|
||||||
r.ReadsBusy = node.Stated && node.Contract >= BusyContract
|
r.ReadsBusy = node.Stated && node.Contract >= BusyContract
|
||||||
|
|||||||
@@ -35,10 +35,22 @@ func placing(t *testing.T, node string, modules ...string) (string, []byte) {
|
|||||||
for _, m := range modules {
|
for _, m := range modules {
|
||||||
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
||||||
}
|
}
|
||||||
key, err := delivery.PlacementKey(node)
|
values, err := delivery.PlacementValues(p)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
key, _ := delivery.PlacementNodeKey(node)
|
||||||
|
return key, values[key]
|
||||||
|
}
|
||||||
|
|
||||||
|
// placingLegacy is placing in the form before, `placement.<node>` (research 072 row 37).
|
||||||
|
func placingLegacy(t *testing.T, node string, modules ...string) (string, []byte) {
|
||||||
|
t.Helper()
|
||||||
|
p := delivery.Placement{Node: node}
|
||||||
|
for _, m := range modules {
|
||||||
|
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
||||||
|
}
|
||||||
|
key, _ := delivery.LegacyPlacementKey(node)
|
||||||
value, err := delivery.EncodePlacement(p)
|
value, err := delivery.EncodePlacement(p)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
@@ -177,8 +189,21 @@ func TestToolchainGate(t *testing.T) {
|
|||||||
|
|
||||||
t.Run("the history: a module in the first revision heard of a key stands since before it", func(t *testing.T) {
|
t.Run("the history: a module in the first revision heard of a key stands since before it", func(t *testing.T) {
|
||||||
p := delivery.NewPlaced()
|
p := delivery.NewPlaced()
|
||||||
k, v := placing(t, "ace", NodeBuildModule)
|
// ace in the form before, then converted: its module stands since before, the legacy key read no more.
|
||||||
|
k, v := placingLegacy(t, "ace", NodeBuildModule)
|
||||||
p.Put(k, v, false, old)
|
p.Put(k, v, false, old)
|
||||||
|
k, v = placing(t, "ace", NodeBuildModule)
|
||||||
|
if _, changed, err := p.Put(k, v, false, placed); len(changed) != 0 || err != nil {
|
||||||
|
t.Errorf("the conversion of ace's key changed %v, %v", changed, err)
|
||||||
|
}
|
||||||
|
if _, changed, _ := p.Put("ace."+NodeBuildModule, []byte("{}"), false, placed); len(changed) != 0 {
|
||||||
|
t.Errorf("a module key says which modules stand: %v", changed)
|
||||||
|
}
|
||||||
|
if k, v := placingLegacy(t, "ace"); true {
|
||||||
|
if _, changed, _ := p.Put(k, v, false, placed); len(changed) != 0 {
|
||||||
|
t.Errorf("the legacy key is read after the node key: %v", changed)
|
||||||
|
}
|
||||||
|
}
|
||||||
k2, v2 := placing(t, "g14")
|
k2, v2 := placing(t, "g14")
|
||||||
p.Put(k2, v2, false, old)
|
p.Put(k2, v2, false, old)
|
||||||
k2, v2 = placing(t, "g14", NodeBuildModule)
|
k2, v2 = placing(t, "g14", NodeBuildModule)
|
||||||
|
|||||||
@@ -86,7 +86,7 @@ func TestTheControlRecordsReadTheirWritersValues(t *testing.T) {
|
|||||||
if _, err := ModuleNames(map[string][]byte{"a.b": nil}); err == nil {
|
if _, err := ModuleNames(map[string][]byte{"a.b": nil}); err == nil {
|
||||||
t.Error("a key that is no module's was read as one")
|
t.Error("a key that is no module's was read as one")
|
||||||
}
|
}
|
||||||
if n := MachineNames(map[string][]byte{"placement.novox": nil, "placement.ace": nil, "proposal.p1": nil}); strings.Join(n, ",") != "ace,novox" {
|
if n := MachineNames(map[string][]byte{"placement.novox": nil, "placement.ace": nil, "ace.node": nil, "ace.app": nil, "g14.app": nil, "proposal.p1": nil}); strings.Join(n, ",") != "ace,novox" {
|
||||||
t.Errorf("the machines are %v", n)
|
t.Errorf("the machines are %v", n)
|
||||||
}
|
}
|
||||||
if h, _ := HoldersOf(keys, "node-files", "novox"); h != nil {
|
if h, _ := HoldersOf(keys, "node-files", "novox"); h != nil {
|
||||||
|
|||||||
@@ -264,11 +264,14 @@ func ModuleNames(keys map[string][]byte) ([]string, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// MachineNames is every machine the placement seat places modules on, sorted, from its state `placement` as a reader
|
// MachineNames is every machine the placement seat places modules on, sorted, from its state `placement` as a reader
|
||||||
// holds its keys: each `placement.<node>`. Keys of the state's other kinds (a proposal's) are passed over.
|
// holds its keys: each with a node key `<node>.node` or, in the form before, `placement.<node>` (placement_keys.go).
|
||||||
|
// Module keys and keys of the state's other kinds (a proposal's) are passed over.
|
||||||
func MachineNames(keys map[string][]byte) []string {
|
func MachineNames(keys map[string][]byte) []string {
|
||||||
|
seen := map[string]bool{}
|
||||||
out := []string{}
|
out := []string{}
|
||||||
for k := range keys {
|
for k := range keys {
|
||||||
if node, err := ParsePlacementKey(k); err == nil {
|
if node, _, kind, err := ParsePlacementKey(k); err == nil && kind != PlacementModuleKind && !seen[node] {
|
||||||
|
seen[node] = true
|
||||||
out = append(out, node)
|
out = append(out, node)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+38
-7
@@ -14,7 +14,9 @@ import (
|
|||||||
// with the controller's lease, ADR 0549 §6).
|
// with the controller's lease, ADR 0549 §6).
|
||||||
//
|
//
|
||||||
// Its input is the state's revisions in the order the bus wrote them, as a watch with the history (jetstream's
|
// Its input is the state's revisions in the order the bus wrote them, as a watch with the history (jetstream's
|
||||||
// IncludeHistory) hands them: Put takes each. A module stands since the revision of its node's key that first named it
|
// IncludeHistory) hands them: Put takes each, in either form of the state's keys (placement_keys.go): a node's node key
|
||||||
|
// names its modules (a module key says nothing of which stand), and its legacy key `placement.<node>` does until a node
|
||||||
|
// key of it is heard. A module stands since the revision of its node's key that first named it
|
||||||
// in an unbroken run. A module in the first revision heard of a key stands since the zero time: that revision is the
|
// in an unbroken run. A module in the first revision heard of a key stands since the zero time: that revision is the
|
||||||
// key's first (the seed of ADR 0549 §4 placed what already ran there) or the oldest the bus still keeps
|
// key's first (the seed of ADR 0549 §4 placed what already ran there) or the oldest the bus still keeps
|
||||||
// (PlacementHistory), and either way the module may have run there before it. A key's value that does not read places
|
// (PlacementHistory), and either way the module may have run there before it. A key's value that does not read places
|
||||||
@@ -25,31 +27,60 @@ type Placed struct {
|
|||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
since map[string]map[string]time.Time // node -> module -> since when it stands
|
since map[string]map[string]time.Time // node -> module -> since when it stands
|
||||||
heard map[string]bool // the nodes whose key was heard at all
|
heard map[string]bool // the nodes whose key was heard at all
|
||||||
|
// newForm are the nodes whose node key `<node>.node` was heard: their legacy key `placement.<node>` is no longer
|
||||||
|
// read (placement_keys.go).
|
||||||
|
newForm map[string]bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewPlaced is a Placed that holds no placement yet.
|
// NewPlaced is a Placed that holds no placement yet.
|
||||||
func NewPlaced() *Placed {
|
func NewPlaced() *Placed {
|
||||||
return &Placed{since: map[string]map[string]time.Time{}, heard: map[string]bool{}}
|
return &Placed{since: map[string]map[string]time.Time{}, heard: map[string]bool{}, newForm: map[string]bool{}}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Put takes one revision of a key of the state `placement` (its value, whether it is a delete, and when the bus wrote
|
// Put takes one revision of a key of the state `placement` (its value, whether it is a delete, and when the bus wrote
|
||||||
// it). It answers the node the key is of and the modules whose standing on it changed (placed or removed), in name
|
// it). It answers the node the key is of and the modules whose standing on it changed (placed or removed), in name
|
||||||
// order, and an error for a key or a value that does not read (the node then places nothing).
|
// order, and an error for a key or a value that does not read (the node then places nothing).
|
||||||
func (p *Placed) Put(key string, value []byte, deleted bool, at time.Time) (node string, changed []string, err error) {
|
func (p *Placed) Put(key string, value []byte, deleted bool, at time.Time) (node string, changed []string, err error) {
|
||||||
node, err = ParsePlacementKey(key)
|
node, _, kind, err := ParsePlacementKey(key)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", nil, err
|
return "", nil, err
|
||||||
}
|
}
|
||||||
|
if kind == PlacementModuleKind {
|
||||||
|
return node, nil, nil // which modules stand is the node key's to say
|
||||||
|
}
|
||||||
|
p.mu.Lock()
|
||||||
|
if kind == PlacementLegacyKind && p.newForm[node] {
|
||||||
|
p.mu.Unlock()
|
||||||
|
return node, nil, nil // the node is read from its node key once one was heard
|
||||||
|
}
|
||||||
|
if kind == PlacementNodeKind {
|
||||||
|
p.newForm[node] = true
|
||||||
|
}
|
||||||
|
p.mu.Unlock()
|
||||||
now := map[string]bool{}
|
now := map[string]bool{}
|
||||||
if !deleted {
|
if !deleted {
|
||||||
pl, derr := DecodePlacement(value)
|
var modules []string
|
||||||
|
var derr error
|
||||||
|
var named string
|
||||||
|
if kind == PlacementNodeKind {
|
||||||
|
var n PlacementNode
|
||||||
|
n, derr = DecodePlacementNode(value)
|
||||||
|
named = n.Node
|
||||||
|
for _, r := range n.Modules {
|
||||||
|
modules = append(modules, r.Module)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
var pl Placement
|
||||||
|
pl, derr = DecodePlacement(value)
|
||||||
|
named, modules = pl.Node, pl.Assigned()
|
||||||
|
}
|
||||||
switch {
|
switch {
|
||||||
case derr != nil:
|
case derr != nil:
|
||||||
err = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr)
|
err = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr)
|
||||||
case pl.Node != node:
|
case named != node:
|
||||||
err = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, pl.Node)
|
err = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, named)
|
||||||
default:
|
default:
|
||||||
for _, m := range pl.Assigned() {
|
for _, m := range modules {
|
||||||
now[m] = true
|
now[m] = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -11,8 +11,9 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// Placement of modules on nodes is the state of the seat mesh-placement (novox/hq ADR 0567), held once for the mesh by
|
// Placement of modules on nodes is the state of the seat mesh-placement (novox/hq ADR 0567), held once for the mesh by
|
||||||
// the assigner: one key `placement.<node>` per node in its state `placement`, written only on the operator's Approve of a
|
// the assigner: in its state `placement`, one node key `<node>.node` and one key `<node>.<module>` per module placed
|
||||||
// proposal the nodes it touches answered (§2-§5), by compare-and-set at the revision the proposal was laid over. No
|
// (placement_keys.go; research 072 row 37), written only on the operator's Approve of a proposal the nodes it touches
|
||||||
|
// answered (§2-§5), by compare-and-set at the revision the proposal was laid over. No
|
||||||
// repository holds it. The node agents, the grants part, mesh-config and the status view read it (§9); nothing in it is
|
// repository holds it. The node agents, the grants part, mesh-config and the status view read it (§9); nothing in it is
|
||||||
// secret. Its states are named after the holding module, as every module state is, and these constants are the seat's
|
// secret. Its states are named after the holding module, as every module state is, and these constants are the seat's
|
||||||
// contract: every reader takes the names from here (§1).
|
// contract: every reader takes the names from here (§1).
|
||||||
@@ -42,23 +43,6 @@ const (
|
|||||||
// PlacementEvents are the seat's four events, in the order a proposal says them.
|
// PlacementEvents are the seat's four events, in the order a proposal says them.
|
||||||
var PlacementEvents = []string{PlacementProposedEvent, PlacementJudgedEvent, PlacementChangedEvent, PlacementWithdrawnEvent}
|
var PlacementEvents = []string{PlacementProposedEvent, PlacementJudgedEvent, PlacementChangedEvent, PlacementWithdrawnEvent}
|
||||||
|
|
||||||
// PlacementKey is the key, in PlacementBucket, of one node's placement: `placement.<node>`.
|
|
||||||
func PlacementKey(node string) (string, error) {
|
|
||||||
if err := checkName("node", node); err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
return placementWord + "." + node, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// ParsePlacementKey is the inverse of PlacementKey; it refuses any other key, naming it.
|
|
||||||
func ParsePlacementKey(key string) (node string, err error) {
|
|
||||||
node, ok := strings.CutPrefix(key, placementWord+".")
|
|
||||||
if !ok || checkName("node", node) != nil {
|
|
||||||
return "", fmt.Errorf("delivery: %q is not a placement key, placement.<node>", key)
|
|
||||||
}
|
|
||||||
return node, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Provider is the module on a node that answers a requirement or holds a seat: a pin's or a binding's provider.
|
// Provider is the module on a node that answers a requirement or holds a seat: a pin's or a binding's provider.
|
||||||
type Provider struct {
|
type Provider struct {
|
||||||
Node string `json:"node"`
|
Node string `json:"node"`
|
||||||
|
|||||||
@@ -0,0 +1,596 @@
|
|||||||
|
package delivery
|
||||||
|
|
||||||
|
import (
|
||||||
|
"crypto/sha256"
|
||||||
|
"encoding/hex"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"sort"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
)
|
||||||
|
|
||||||
|
// **One key per module and node** (research 072 row 37, the operator's decision of 2026-10-31; issue 774). The seat
|
||||||
|
// mesh-placement's state `placement` keeps a node's placement under the node's own keys, so a node watches `<node>.>`
|
||||||
|
// alone and a change of one module writes that module's key:
|
||||||
|
//
|
||||||
|
// - `<node>.node`, the node key: what a person decided of the node (NodeDecided), its pins and its holdings, and the
|
||||||
|
// modules placed on it in the order the composition reads them, each with the digest of its module key's value.
|
||||||
|
// The word `node` is the have state's own word for a node's key (HaveNodeWord), so no module is named it.
|
||||||
|
// - `<node>.<module>`, a module key: the module's ports, bindings and layers (PlacedModule).
|
||||||
|
//
|
||||||
|
// The node key is the commit of a write (ADR 0567 §6: a proposal's keys share one revision in history). A writer writes
|
||||||
|
// the module keys first, each carrying the value it replaces as Previous, then the node key, by compare-and-set at the
|
||||||
|
// revision the proposal was laid over, and deletes the module keys no longer named last. A reader takes, for each module
|
||||||
|
// the node key names, the value of its module key whose digest the node key names: the current one once the node key
|
||||||
|
// is written, the previous one while the write is under way. So a reader never composes a part of a write, and every
|
||||||
|
// write of a node is one revision of its node key, whose history is the node's placement history as `placement.<node>`'s
|
||||||
|
// was (Placed, the warrant's record). A module key the node key does not name is not placed.
|
||||||
|
//
|
||||||
|
// The form before, one value `placement.<node>` per node (LegacyPlacementKey), is still read during the delivery of
|
||||||
|
// this change, so readers and the holder may move in any order: a node with a node key is read from its new keys alone,
|
||||||
|
// one without from its legacy key. The holder converts each legacy value once at its start (PlacementKeys.Migrations),
|
||||||
|
// keeps the legacy key in step after every commit as a mirror, so a reader of the form before alone still follows every
|
||||||
|
// write while the members of the delivery move in any order, and deletes it first when a node's placement is deleted,
|
||||||
|
// so a placement handed back never returns from it. A later change stops the mirror and deletes `placement.*`.
|
||||||
|
const (
|
||||||
|
// PlacementNodeWord is the last word of a node key, `<node>.node`.
|
||||||
|
PlacementNodeWord = HaveNodeWord
|
||||||
|
placementDigestLen = 16
|
||||||
|
)
|
||||||
|
|
||||||
|
// PlacementKeyKind is the kind of a key of the state `placement`.
|
||||||
|
type PlacementKeyKind int
|
||||||
|
|
||||||
|
// The kinds of key: a node key, a module key, and the form before (one value per node).
|
||||||
|
const (
|
||||||
|
PlacementNodeKind PlacementKeyKind = iota + 1
|
||||||
|
PlacementModuleKind
|
||||||
|
PlacementLegacyKind
|
||||||
|
)
|
||||||
|
|
||||||
|
// PlacementNodeKey is a node's node key, `<node>.node`. A node named `placement` is refused: its keys would read as the
|
||||||
|
// legacy form's.
|
||||||
|
func PlacementNodeKey(node string) (string, error) {
|
||||||
|
if err := checkPlacementNode(node); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return node + "." + PlacementNodeWord, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PlacementModuleKey is a module's key on a node, `<node>.<module>`; a module named `node` is refused.
|
||||||
|
func PlacementModuleKey(node, module string) (string, error) {
|
||||||
|
if err := checkPlacementNode(node); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
if err := checkPlacementModule(module); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return node + "." + module, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PlacementNodeKeys is the pattern of every key of one node, `<node>.>`: what a node watches of its own placement.
|
||||||
|
func PlacementNodeKeys(node string) (string, error) {
|
||||||
|
if err := checkPlacementNode(node); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return node + ".>", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// LegacyPlacementKey is a node's key in the form before, `placement.<node>`, read while this change is delivered.
|
||||||
|
func LegacyPlacementKey(node string) (string, error) {
|
||||||
|
if err := checkName("node", node); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return placementWord + "." + node, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// ParsePlacementKey reads a key of the state `placement` in either form: its node, its module (empty for a node key and
|
||||||
|
// a legacy key) and its kind. Any other key is refused, naming it.
|
||||||
|
func ParsePlacementKey(key string) (node, module string, kind PlacementKeyKind, err error) {
|
||||||
|
first, rest, ok := strings.Cut(key, ".")
|
||||||
|
if !ok || strings.Contains(rest, ".") || checkName("node", first) != nil || checkName("module", rest) != nil {
|
||||||
|
return "", "", 0, fmt.Errorf("delivery: %q is not a placement key, <node>.node or <node>.<module>", key)
|
||||||
|
}
|
||||||
|
switch {
|
||||||
|
case first == placementWord:
|
||||||
|
return rest, "", PlacementLegacyKind, nil
|
||||||
|
case rest == PlacementNodeWord:
|
||||||
|
return first, "", PlacementNodeKind, nil
|
||||||
|
default:
|
||||||
|
return first, rest, PlacementModuleKind, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func checkPlacementNode(node string) error {
|
||||||
|
if err := checkName("node", node); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if node == placementWord {
|
||||||
|
return fmt.Errorf("delivery: no node is named %q: its placement keys would read as the form before", placementWord)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func checkPlacementModule(module string) error {
|
||||||
|
if err := checkName("module", module); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if module == PlacementNodeWord {
|
||||||
|
return fmt.Errorf("delivery: no module is named %q: it is the word of a node's own key", PlacementNodeWord)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PlacementRef is one module the node key names, with the digest of the module key's value it stands for.
|
||||||
|
type PlacementRef struct {
|
||||||
|
Module string `json:"module"`
|
||||||
|
Digest string `json:"digest"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// PlacementNode is the value of a node key: the node's decisions, pins and holdings, and the modules placed on it.
|
||||||
|
type PlacementNode struct {
|
||||||
|
V int `json:"v"`
|
||||||
|
Node string `json:"node"`
|
||||||
|
Decided NodeDecided `json:"decided"`
|
||||||
|
Modules []PlacementRef `json:"modules"`
|
||||||
|
Pins map[string]Provider `json:"pins,omitempty"`
|
||||||
|
Holds []Holding `json:"holds,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// PlacementModule is the value of a module key: the module as placed, and, while a write that replaces it is under way,
|
||||||
|
// the value it replaces (Previous), which a reader takes until the node key names the new one.
|
||||||
|
type PlacementModule struct {
|
||||||
|
V int `json:"v"`
|
||||||
|
Node string `json:"node"`
|
||||||
|
Placed PlacedModule `json:"placed"`
|
||||||
|
Previous *PlacedModule `json:"previous,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Digest is the digest of a placed module's value, as a node key names it.
|
||||||
|
func (m PlacedModule) Digest() string {
|
||||||
|
raw, _ := json.Marshal(m)
|
||||||
|
sum := sha256.Sum256(raw)
|
||||||
|
return hex.EncodeToString(sum[:])[:placementDigestLen]
|
||||||
|
}
|
||||||
|
|
||||||
|
// EncodePlacementNode is n as JSON at Version.
|
||||||
|
func EncodePlacementNode(n PlacementNode) ([]byte, error) {
|
||||||
|
if err := checkPlacementNode(n.Node); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
n.V = Version
|
||||||
|
return json.Marshal(n)
|
||||||
|
}
|
||||||
|
|
||||||
|
// DecodePlacementNode reads a node key's value; it refuses an unknown version, a node or module that is not a name, and
|
||||||
|
// a module named twice or without a digest.
|
||||||
|
func DecodePlacementNode(data []byte) (PlacementNode, error) {
|
||||||
|
var n PlacementNode
|
||||||
|
if err := decode(data, &n); err != nil {
|
||||||
|
return PlacementNode{}, err
|
||||||
|
}
|
||||||
|
if err := checkPlacementNode(n.Node); err != nil {
|
||||||
|
return PlacementNode{}, err
|
||||||
|
}
|
||||||
|
seen := map[string]bool{}
|
||||||
|
for _, r := range n.Modules {
|
||||||
|
if err := checkPlacementModule(r.Module); err != nil {
|
||||||
|
return PlacementNode{}, err
|
||||||
|
}
|
||||||
|
if seen[r.Module] || r.Digest == "" {
|
||||||
|
return PlacementNode{}, fmt.Errorf("delivery: the node key of %s names %s twice or without a digest", n.Node, r.Module)
|
||||||
|
}
|
||||||
|
seen[r.Module] = true
|
||||||
|
}
|
||||||
|
return n, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// EncodePlacementModule is m as JSON at Version.
|
||||||
|
func EncodePlacementModule(m PlacementModule) ([]byte, error) {
|
||||||
|
if err := checkPlacementNode(m.Node); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := checkPlacementModule(m.Placed.Module); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if m.Previous != nil && m.Previous.Module != m.Placed.Module {
|
||||||
|
return nil, fmt.Errorf("delivery: the key of %s on %s carries %s as its previous value", m.Placed.Module, m.Node, m.Previous.Module)
|
||||||
|
}
|
||||||
|
m.V = Version
|
||||||
|
return json.Marshal(m)
|
||||||
|
}
|
||||||
|
|
||||||
|
// DecodePlacementModule reads a module key's value.
|
||||||
|
func DecodePlacementModule(data []byte) (PlacementModule, error) {
|
||||||
|
var m PlacementModule
|
||||||
|
if err := decode(data, &m); err != nil {
|
||||||
|
return PlacementModule{}, err
|
||||||
|
}
|
||||||
|
if err := checkPlacementNode(m.Node); err != nil {
|
||||||
|
return PlacementModule{}, err
|
||||||
|
}
|
||||||
|
if err := checkPlacementModule(m.Placed.Module); err != nil {
|
||||||
|
return PlacementModule{}, err
|
||||||
|
}
|
||||||
|
return m, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PlacementKeys is the state `placement` as a reader holds it, key by key in either form, and every node's placement
|
||||||
|
// assembled from it: the one reader of the state's keys for every part (issue 774), so no part parses them itself. A
|
||||||
|
// write of the holder's under way is never assembled half: a module key's value is taken only as far as the node key
|
||||||
|
// names its digest (see the package's comment above PlacementNodeWord), and a node whose keys do not agree keeps the
|
||||||
|
// placement last assembled for it, Put saying why.
|
||||||
|
//
|
||||||
|
// Safe for use from a watch and from a reader at once.
|
||||||
|
type PlacementKeys struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
nodes map[string]*placementKeysOf
|
||||||
|
}
|
||||||
|
|
||||||
|
// placementKeysOf is what is held of one node's keys.
|
||||||
|
type placementKeysOf struct {
|
||||||
|
node *PlacementNode
|
||||||
|
nodeRev uint64
|
||||||
|
modules map[string]PlacementModule
|
||||||
|
moduleRevs map[string]uint64
|
||||||
|
legacy *Placement
|
||||||
|
legacyRev uint64
|
||||||
|
last *Placement // the placement last assembled whole, nil when none
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewPlacementKeys holds no key yet.
|
||||||
|
func NewPlacementKeys() *PlacementKeys { return &PlacementKeys{nodes: map[string]*placementKeysOf{}} }
|
||||||
|
|
||||||
|
// PlacementsOf is every node's placement assembled from the state's keys as a reader holds them (key to value; a
|
||||||
|
// deleted key absent), and an error naming every key that did not read and every node whose keys do not agree (that
|
||||||
|
// node is then assembled from its module keys' values as they are).
|
||||||
|
func PlacementsOf(keys map[string][]byte) (map[string]Placement, error) {
|
||||||
|
k := NewPlacementKeys()
|
||||||
|
var errs []error
|
||||||
|
names := make([]string, 0, len(keys))
|
||||||
|
for key := range keys {
|
||||||
|
names = append(names, key)
|
||||||
|
}
|
||||||
|
sort.Strings(names)
|
||||||
|
for _, key := range names {
|
||||||
|
// A node key read before its module keys does not agree yet; whether the keys agree is asked once all are in.
|
||||||
|
if _, readErr, _ := k.put(key, keys[key], 0, false); readErr != nil {
|
||||||
|
errs = append(errs, readErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
k.mu.Lock()
|
||||||
|
for _, node := range sortedKeys(k.nodes) {
|
||||||
|
if _, _, err := k.nodes[node].assemble(node); err != nil {
|
||||||
|
errs = append(errs, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
k.mu.Unlock()
|
||||||
|
return k.All(), errors.Join(errs...)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Put takes one key's value in (revision: the bus's revision of it, 0 when the reader has none; deleted: a delete or a
|
||||||
|
// purge). It answers the node the key is of and an error for a key or a value that does not read, or for a node whose
|
||||||
|
// keys do not agree; the node then keeps what it had.
|
||||||
|
func (k *PlacementKeys) Put(key string, value []byte, revision uint64, deleted bool) (string, error) {
|
||||||
|
node, readErr, err := k.put(key, value, revision, deleted)
|
||||||
|
if readErr != nil {
|
||||||
|
return node, readErr
|
||||||
|
}
|
||||||
|
return node, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// put is Put telling an error of the key or value it took (readErr) from the node's keys not agreeing after it (err).
|
||||||
|
func (k *PlacementKeys) put(key string, value []byte, revision uint64, deleted bool) (node string, readErr, err error) {
|
||||||
|
node, module, kind, err := ParsePlacementKey(key)
|
||||||
|
if err != nil {
|
||||||
|
return "", err, nil
|
||||||
|
}
|
||||||
|
k.mu.Lock()
|
||||||
|
defer k.mu.Unlock()
|
||||||
|
of := k.nodes[node]
|
||||||
|
if of == nil {
|
||||||
|
of = &placementKeysOf{modules: map[string]PlacementModule{}, moduleRevs: map[string]uint64{}}
|
||||||
|
k.nodes[node] = of
|
||||||
|
}
|
||||||
|
switch kind {
|
||||||
|
case PlacementNodeKind:
|
||||||
|
of.node, of.nodeRev = nil, 0
|
||||||
|
if !deleted {
|
||||||
|
n, derr := DecodePlacementNode(value)
|
||||||
|
switch {
|
||||||
|
case derr != nil:
|
||||||
|
readErr = fmt.Errorf("the node key %s does not read, so it places nothing there: %w", key, derr)
|
||||||
|
case n.Node != node:
|
||||||
|
readErr = fmt.Errorf("the node key %s names the node %q, so it places nothing there", key, n.Node)
|
||||||
|
default:
|
||||||
|
of.node, of.nodeRev = &n, revision
|
||||||
|
}
|
||||||
|
}
|
||||||
|
case PlacementModuleKind:
|
||||||
|
delete(of.modules, module)
|
||||||
|
delete(of.moduleRevs, module)
|
||||||
|
if !deleted {
|
||||||
|
m, derr := DecodePlacementModule(value)
|
||||||
|
switch {
|
||||||
|
case derr != nil:
|
||||||
|
readErr = fmt.Errorf("the module key %s does not read: %w", key, derr)
|
||||||
|
case m.Node != node || m.Placed.Module != module:
|
||||||
|
readErr = fmt.Errorf("the module key %s holds %s on %s", key, m.Placed.Module, m.Node)
|
||||||
|
default:
|
||||||
|
of.modules[module] = m
|
||||||
|
}
|
||||||
|
// Its revision is kept even when its value does not read, so the holder can write over it.
|
||||||
|
of.moduleRevs[module] = revision
|
||||||
|
}
|
||||||
|
case PlacementLegacyKind:
|
||||||
|
of.legacy, of.legacyRev = nil, 0
|
||||||
|
if !deleted {
|
||||||
|
p, derr := DecodePlacement(value)
|
||||||
|
switch {
|
||||||
|
case derr != nil:
|
||||||
|
readErr = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr)
|
||||||
|
case p.Node != node:
|
||||||
|
readErr = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, p.Node)
|
||||||
|
default:
|
||||||
|
of.legacy, of.legacyRev = &p, revision
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
p, whole, err := of.assemble(node)
|
||||||
|
switch {
|
||||||
|
case p == nil:
|
||||||
|
of.last = nil
|
||||||
|
case whole || of.last == nil:
|
||||||
|
of.last = p
|
||||||
|
}
|
||||||
|
return node, readErr, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// assemble is the node's placement from its keys: nil when none places it, whole false (and an error) when the node key
|
||||||
|
// names a module digest no module key's value has.
|
||||||
|
func (of *placementKeysOf) assemble(node string) (*Placement, bool, error) {
|
||||||
|
if of.node == nil {
|
||||||
|
if of.legacy == nil {
|
||||||
|
return nil, true, nil
|
||||||
|
}
|
||||||
|
p := clonePlacement(*of.legacy)
|
||||||
|
return &p, true, nil
|
||||||
|
}
|
||||||
|
n := of.node
|
||||||
|
p := Placement{V: Version, Node: node, Decided: n.Decided, Pins: n.Pins, Holds: n.Holds, Modules: make([]PlacedModule, 0, len(n.Modules))}
|
||||||
|
var astray []string
|
||||||
|
for _, r := range n.Modules {
|
||||||
|
m, ok := of.modules[r.Module]
|
||||||
|
switch {
|
||||||
|
case ok && m.Placed.Digest() == r.Digest:
|
||||||
|
p.Modules = append(p.Modules, m.Placed)
|
||||||
|
case ok && m.Previous != nil && m.Previous.Digest() == r.Digest:
|
||||||
|
p.Modules = append(p.Modules, *m.Previous)
|
||||||
|
case ok:
|
||||||
|
astray = append(astray, r.Module)
|
||||||
|
p.Modules = append(p.Modules, m.Placed)
|
||||||
|
default:
|
||||||
|
astray = append(astray, r.Module)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(astray) > 0 {
|
||||||
|
return &p, false, fmt.Errorf("the node key of %s names %s at a value no module key holds: %s keeps the placement it had",
|
||||||
|
node, strings.Join(astray, ", "), node)
|
||||||
|
}
|
||||||
|
return &p, true, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Of is node's placement and the revision of the key it was read from (its node key, or its legacy key), and whether
|
||||||
|
// any key places it.
|
||||||
|
func (k *PlacementKeys) Of(node string) (Placement, uint64, bool) {
|
||||||
|
k.mu.Lock()
|
||||||
|
defer k.mu.Unlock()
|
||||||
|
of := k.nodes[node]
|
||||||
|
if of == nil || of.last == nil {
|
||||||
|
return Placement{}, 0, false
|
||||||
|
}
|
||||||
|
rev := of.nodeRev
|
||||||
|
if of.node == nil {
|
||||||
|
rev = of.legacyRev
|
||||||
|
}
|
||||||
|
return clonePlacement(*of.last), rev, true
|
||||||
|
}
|
||||||
|
|
||||||
|
// All is every node's placement, by node, and the revision of each one's key (Of).
|
||||||
|
func (k *PlacementKeys) All() map[string]Placement {
|
||||||
|
out, _ := k.AllRevisions()
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// AllRevisions is All with the revision of each node's key.
|
||||||
|
func (k *PlacementKeys) AllRevisions() (map[string]Placement, map[string]uint64) {
|
||||||
|
k.mu.Lock()
|
||||||
|
nodes := make([]string, 0, len(k.nodes))
|
||||||
|
for n := range k.nodes {
|
||||||
|
nodes = append(nodes, n)
|
||||||
|
}
|
||||||
|
k.mu.Unlock()
|
||||||
|
out, revs := map[string]Placement{}, map[string]uint64{}
|
||||||
|
for _, n := range nodes {
|
||||||
|
if p, rev, ok := k.Of(n); ok {
|
||||||
|
out[n], revs[n] = p, rev
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out, revs
|
||||||
|
}
|
||||||
|
|
||||||
|
// PlacementKeyWrite is one write of a key of the state `placement`, by compare-and-set: Value nil deletes the key at
|
||||||
|
// Base, Base 0 creates it, any other Base updates it at that revision.
|
||||||
|
type PlacementKeyWrite struct {
|
||||||
|
Node string
|
||||||
|
Key string
|
||||||
|
Value []byte
|
||||||
|
Base uint64
|
||||||
|
// Commit says this write is the node key's: once it is written, the node's placement is the new one.
|
||||||
|
Commit bool
|
||||||
|
// Mirror says this write is the legacy key's, `placement.<node>`, kept in step after the commit for the readers not
|
||||||
|
// yet delivered: a reader of both forms never reads it once the node key is there, so a mirror write that fails
|
||||||
|
// harms no such reader, and is made again at the node's next write.
|
||||||
|
Mirror bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// Writes are the writes that take node from the placement its keys hold to after (nil: no placement), in the order
|
||||||
|
// they are made: the module keys that change, each carrying the value it replaces; then the node key, the commit; then
|
||||||
|
// the legacy key `placement.<node>` with the whole placement, a mirror for the readers not yet delivered, unless it
|
||||||
|
// holds it already; then the deletes of the module keys after no longer names. For no placement the legacy key goes
|
||||||
|
// first, then the node key, then the module keys, so a crash between them never leaves the legacy value to be read
|
||||||
|
// again. A module key whose value stays is not written.
|
||||||
|
func (k *PlacementKeys) Writes(node string, after *Placement) ([]PlacementKeyWrite, error) {
|
||||||
|
nodeKey, err := PlacementNodeKey(node)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
k.mu.Lock()
|
||||||
|
defer k.mu.Unlock()
|
||||||
|
of := k.nodes[node]
|
||||||
|
if of == nil {
|
||||||
|
of = &placementKeysOf{modules: map[string]PlacementModule{}, moduleRevs: map[string]uint64{}}
|
||||||
|
}
|
||||||
|
var out []PlacementKeyWrite
|
||||||
|
deleteModules := func(keep map[string]bool) {
|
||||||
|
for _, module := range sortedKeys(of.moduleRevs) {
|
||||||
|
if keep[module] {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
key, _ := PlacementModuleKey(node, module)
|
||||||
|
out = append(out, PlacementKeyWrite{Node: node, Key: key, Base: of.moduleRevs[module]})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if after == nil {
|
||||||
|
if of.legacy != nil {
|
||||||
|
key, _ := LegacyPlacementKey(node)
|
||||||
|
out = append(out, PlacementKeyWrite{Node: node, Key: key, Base: of.legacyRev})
|
||||||
|
}
|
||||||
|
if of.node != nil {
|
||||||
|
out = append(out, PlacementKeyWrite{Node: node, Key: nodeKey, Base: of.nodeRev, Commit: true})
|
||||||
|
}
|
||||||
|
deleteModules(nil)
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
if after.Node != node {
|
||||||
|
return nil, fmt.Errorf("delivery: the placement written under %s names the node %q", node, after.Node)
|
||||||
|
}
|
||||||
|
if err := after.check(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
committed := map[string]PlacedModule{}
|
||||||
|
if of.last != nil {
|
||||||
|
for _, m := range of.last.Modules {
|
||||||
|
committed[m.Module] = m
|
||||||
|
}
|
||||||
|
}
|
||||||
|
n := PlacementNode{Node: node, Decided: after.Decided, Pins: after.Pins, Holds: after.Holds, Modules: make([]PlacementRef, 0, len(after.Modules))}
|
||||||
|
keep := map[string]bool{}
|
||||||
|
for _, m := range after.Modules {
|
||||||
|
if err := checkPlacementModule(m.Module); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
digest := m.Digest()
|
||||||
|
n.Modules = append(n.Modules, PlacementRef{Module: m.Module, Digest: digest})
|
||||||
|
keep[m.Module] = true
|
||||||
|
was, wasPlaced := committed[m.Module]
|
||||||
|
if held, ok := of.modules[m.Module]; ok && held.Placed.Digest() == digest &&
|
||||||
|
(held.Previous == nil || wasPlaced && was.Digest() == digest) {
|
||||||
|
continue // the key holds it, and no previous value a reader could take in its place
|
||||||
|
}
|
||||||
|
v := PlacementModule{Node: node, Placed: m}
|
||||||
|
if wasPlaced && was.Digest() != digest {
|
||||||
|
v.Previous = &was
|
||||||
|
}
|
||||||
|
raw, err := EncodePlacementModule(v)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
key, _ := PlacementModuleKey(node, m.Module)
|
||||||
|
out = append(out, PlacementKeyWrite{Node: node, Key: key, Value: raw, Base: of.moduleRevs[m.Module]})
|
||||||
|
}
|
||||||
|
raw, err := EncodePlacementNode(n)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, PlacementKeyWrite{Node: node, Key: nodeKey, Value: raw, Base: of.nodeRev, Commit: true})
|
||||||
|
// The mirror of the form before, while readers of it alone may run (the delivery of this change); a later change
|
||||||
|
// stops writing it and deletes it.
|
||||||
|
if of.legacy == nil || !SamePlacement(*of.legacy, *after) {
|
||||||
|
legacy, err := EncodePlacement(*after)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
key, _ := LegacyPlacementKey(node)
|
||||||
|
out = append(out, PlacementKeyWrite{Node: node, Key: key, Value: legacy, Base: of.legacyRev, Mirror: true})
|
||||||
|
}
|
||||||
|
deleteModules(keep)
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Migrations are, by node, the writes that convert every node held only in the form before (its legacy key, no node
|
||||||
|
// key) into its node and module keys, its legacy key kept as the mirror for the readers not yet delivered (Writes); empty
|
||||||
|
// once every node is converted, so the holder's conversion at each start is idempotent.
|
||||||
|
func (k *PlacementKeys) Migrations() (map[string][]PlacementKeyWrite, error) {
|
||||||
|
k.mu.Lock()
|
||||||
|
var legacy []string
|
||||||
|
for node, of := range k.nodes {
|
||||||
|
if of.node == nil && of.legacy != nil && checkPlacementNode(node) == nil {
|
||||||
|
legacy = append(legacy, node)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
k.mu.Unlock()
|
||||||
|
sort.Strings(legacy)
|
||||||
|
out := map[string][]PlacementKeyWrite{}
|
||||||
|
var errs []error
|
||||||
|
for _, node := range legacy {
|
||||||
|
p, _, _ := k.Of(node)
|
||||||
|
w, err := k.Writes(node, &p)
|
||||||
|
if err != nil {
|
||||||
|
errs = append(errs, fmt.Errorf("%s: %w", node, err))
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
out[node] = w
|
||||||
|
}
|
||||||
|
return out, errors.Join(errs...)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Legacy says node is held only in the form before.
|
||||||
|
func (k *PlacementKeys) Legacy(node string) bool {
|
||||||
|
k.mu.Lock()
|
||||||
|
defer k.mu.Unlock()
|
||||||
|
of := k.nodes[node]
|
||||||
|
return of != nil && of.node == nil && of.legacy != nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PlacementValues is every key and value of p written whole onto no keys before: its module keys, its node key and the
|
||||||
|
// legacy mirror, as a writer that holds no key of the node (a test's bus, a fresh mesh) writes them.
|
||||||
|
func PlacementValues(p Placement) (map[string][]byte, error) {
|
||||||
|
writes, err := NewPlacementKeys().Writes(p.Node, &p)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out := make(map[string][]byte, len(writes))
|
||||||
|
for _, w := range writes {
|
||||||
|
out[w.Key] = w.Value
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// InsideAddresses is every node's address on the private network as its placement decides it (NodeDecided.Address),
|
||||||
|
// sorted and each once: the machines inside the private network, which a membership's Mesh carried until the assignments
|
||||||
|
// stream went (ADR 0167; research 072 rows 33 and 34). A node with no address decided is left out.
|
||||||
|
func InsideAddresses(placements map[string]Placement) []string {
|
||||||
|
seen := map[string]bool{}
|
||||||
|
out := []string{}
|
||||||
|
for _, p := range placements {
|
||||||
|
if a := strings.TrimSpace(p.Decided.Address); a != "" && !seen[a] {
|
||||||
|
seen[a] = true
|
||||||
|
out = append(out, a)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sort.Strings(out)
|
||||||
|
return out
|
||||||
|
}
|
||||||
@@ -18,20 +18,39 @@ import (
|
|||||||
// a change says, and the catalogue's check of every file.
|
// a change says, and the catalogue's check of every file.
|
||||||
func TestPlacement(t *testing.T) {
|
func TestPlacement(t *testing.T) {
|
||||||
t.Run("key and file", func(t *testing.T) {
|
t.Run("key and file", func(t *testing.T) {
|
||||||
key, err := PlacementKey("g14")
|
// Research 072 row 37: one node key and one key per module, a legacy key read beside them, any other refused.
|
||||||
if err != nil || key != "placement.g14" {
|
for key, want := range map[string]struct {
|
||||||
t.Fatalf("PlacementKey = %q, %v", key, err)
|
node, module string
|
||||||
|
kind PlacementKeyKind
|
||||||
|
}{"g14.node": {"g14", "", PlacementNodeKind}, "g14.app": {"g14", "app", PlacementModuleKind},
|
||||||
|
"placement.g14": {"g14", "", PlacementLegacyKind}} {
|
||||||
|
if n, m, k, err := ParsePlacementKey(key); err != nil || n != want.node || m != want.module || k != want.kind {
|
||||||
|
t.Errorf("ParsePlacementKey(%q) = %q, %q, %d, %v", key, n, m, k, err)
|
||||||
}
|
}
|
||||||
if n, err := ParsePlacementKey(key); err != nil || n != "g14" {
|
|
||||||
t.Fatalf("ParsePlacementKey = %q, %v", n, err)
|
|
||||||
}
|
}
|
||||||
for _, bad := range []string{"want.g14.x", "placement.", "placement.a.b"} {
|
for _, bad := range []string{"want.g14.x", "placement.", "placement.a.b", "g14", "g14.a.b", ".node"} {
|
||||||
if _, err := ParsePlacementKey(bad); err == nil {
|
if _, _, _, err := ParsePlacementKey(bad); err == nil {
|
||||||
t.Errorf("ParsePlacementKey(%q) took it", bad)
|
t.Errorf("ParsePlacementKey(%q) took it", bad)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if _, err := PlacementKey("a.b"); err == nil {
|
if k, err := PlacementNodeKey("g14"); err != nil || k != "g14.node" {
|
||||||
t.Error("PlacementKey took a dotted node")
|
t.Errorf("PlacementNodeKey = %q, %v", k, err)
|
||||||
|
}
|
||||||
|
if k, err := PlacementModuleKey("g14", "app"); err != nil || k != "g14.app" {
|
||||||
|
t.Errorf("PlacementModuleKey = %q, %v", k, err)
|
||||||
|
}
|
||||||
|
if k, err := PlacementNodeKeys("g14"); err != nil || k != "g14.>" {
|
||||||
|
t.Errorf("PlacementNodeKeys = %q, %v", k, err)
|
||||||
|
}
|
||||||
|
if k, err := LegacyPlacementKey("g14"); err != nil || k != "placement.g14" {
|
||||||
|
t.Errorf("LegacyPlacementKey = %q, %v", k, err)
|
||||||
|
}
|
||||||
|
for name, err := range map[string]error{"a dotted node": func() error { _, err := PlacementNodeKey("a.b"); return err }(),
|
||||||
|
"a node named placement": func() error { _, err := PlacementNodeKey("placement"); return err }(),
|
||||||
|
"a module named node": func() error { _, err := PlacementModuleKey("g14", "node"); return err }()} {
|
||||||
|
if err == nil {
|
||||||
|
t.Errorf("%s was taken", name)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
// ADR 0567 §1: the seat's three states are named after the holding module, and every reader takes the names here.
|
// ADR 0567 §1: the seat's three states are named after the holding module, and every reader takes the names here.
|
||||||
for got, want := range map[string]string{PlacementBucket: "mesh-delivery-assigner_placement",
|
for got, want := range map[string]string{PlacementBucket: "mesh-delivery-assigner_placement",
|
||||||
@@ -149,6 +168,127 @@ func TestPlacement(t *testing.T) {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// Research 072 row 37: a node's placement as its node key and module keys, written so a reader never takes a part of
|
||||||
|
// a write, read beside the form before, and converted from it once.
|
||||||
|
t.Run("one key per module and node", func(t *testing.T) {
|
||||||
|
values, err := PlacementValues(g14)
|
||||||
|
if err != nil || len(values) != 4 || values["g14.node"] == nil || values["g14.zsh"] == nil || values["g14.app"] == nil ||
|
||||||
|
values["placement.g14"] == nil {
|
||||||
|
t.Fatalf("PlacementValues = %v, %v", values, err)
|
||||||
|
}
|
||||||
|
got, err := PlacementsOf(values)
|
||||||
|
if err != nil || !SamePlacement(got["g14"], g14) || len(got) != 1 {
|
||||||
|
t.Fatalf("PlacementsOf = %+v, %v", got, err)
|
||||||
|
}
|
||||||
|
if got := InsideAddresses(map[string]Placement{"g14": g14, "novox": novox, "x": {Node: "x"}, "y": {Node: "y", Decided: NodeDecided{Address: " 10.77.0.1"}}}); strings.Join(got, ",") != "10.77.0.1,10.77.0.2" {
|
||||||
|
t.Errorf("InsideAddresses = %v", got)
|
||||||
|
}
|
||||||
|
legacy, _ := EncodePlacement(novox)
|
||||||
|
stale, _ := EncodePlacement(Placement{Node: "g14"})
|
||||||
|
both := map[string][]byte{"placement.novox": legacy, "placement.g14": stale}
|
||||||
|
for k, v := range values {
|
||||||
|
both[k] = v
|
||||||
|
}
|
||||||
|
if got, err := PlacementsOf(both); err != nil || !SamePlacement(got["g14"], g14) || !SamePlacement(got["novox"], novox) {
|
||||||
|
t.Errorf("a node key wins over a legacy key, and a legacy key alone places its node: %+v, %v", got, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
keys := NewPlacementKeys()
|
||||||
|
rev := uint64(0)
|
||||||
|
apply := func(writes []PlacementKeyWrite, upTo int) {
|
||||||
|
t.Helper()
|
||||||
|
for i, w := range writes {
|
||||||
|
if i == upTo {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
rev++
|
||||||
|
if _, err := keys.Put(w.Key, w.Value, rev, w.Value == nil); err != nil {
|
||||||
|
t.Fatalf("write %d (%s): %v", i, w.Key, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// The form before converts once: its node and module keys, the legacy key left for the readers not yet delivered.
|
||||||
|
rev++
|
||||||
|
legacyG14, _ := EncodePlacement(g14)
|
||||||
|
if _, err := keys.Put("placement.g14", legacyG14, rev, false); err != nil || !keys.Legacy("g14") {
|
||||||
|
t.Fatalf("the legacy key: %v", err)
|
||||||
|
}
|
||||||
|
migrations, err := keys.Migrations()
|
||||||
|
if err != nil || len(migrations["g14"]) != 3 || !migrations["g14"][2].Commit {
|
||||||
|
// The legacy key holds the placement already: no mirror write.
|
||||||
|
t.Fatalf("Migrations = %+v, %v", migrations, err)
|
||||||
|
}
|
||||||
|
apply(migrations["g14"], -1)
|
||||||
|
if m, _ := keys.Migrations(); len(m) != 0 || keys.Legacy("g14") {
|
||||||
|
t.Fatalf("a converted node converts again: %+v", m)
|
||||||
|
}
|
||||||
|
if p, r, ok := keys.Of("g14"); !ok || !SamePlacement(p, g14) || r != rev {
|
||||||
|
t.Fatalf("Of after the conversion = %+v, %d, %v", p, r, ok)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A write under way is never read half: app's new port is not read before the node key names it.
|
||||||
|
after := clonePlacement(g14)
|
||||||
|
after.Modules = []PlacedModule{{Module: "app", Ports: map[string]int{"8080": 18081}}, {Module: "new"}}
|
||||||
|
after.Holds, after.Pins = nil, nil
|
||||||
|
writes, err := keys.Writes("g14", &after)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var kinds []string
|
||||||
|
for _, w := range writes {
|
||||||
|
kinds = append(kinds, fmt.Sprintf("%s:%v:%d", w.Key, w.Value != nil, w.Base))
|
||||||
|
}
|
||||||
|
if strings.Join(kinds, " ") != "g14.app:true:3 g14.new:true:0 g14.node:true:4 placement.g14:true:1 g14.zsh:false:2" ||
|
||||||
|
!writes[3].Mirror {
|
||||||
|
t.Fatalf("Writes = %v", kinds)
|
||||||
|
}
|
||||||
|
apply(writes, 2)
|
||||||
|
if p, _, _ := keys.Of("g14"); !SamePlacement(p, g14) {
|
||||||
|
t.Errorf("the module keys written before the node key are read: %+v", p)
|
||||||
|
}
|
||||||
|
apply(writes[2:], -1)
|
||||||
|
if p, _, _ := keys.Of("g14"); !SamePlacement(p, after) {
|
||||||
|
t.Errorf("the write done is not read whole: %+v", p)
|
||||||
|
}
|
||||||
|
if legacy, _ := PlacementsOf(map[string][]byte{"placement.g14": writes[3].Value}); !SamePlacement(legacy["g14"], after) {
|
||||||
|
t.Errorf("the mirror is not the placement written: %+v", legacy)
|
||||||
|
}
|
||||||
|
// A module key that stays is not written again, nor a mirror that holds it.
|
||||||
|
if again, _ := keys.Writes("g14", &after); len(again) != 1 || !again[0].Commit {
|
||||||
|
t.Errorf("an unchanged placement writes %+v", again)
|
||||||
|
}
|
||||||
|
// Keys that do not agree keep what the node had, and say so.
|
||||||
|
stray, _ := EncodePlacementModule(PlacementModule{Node: "g14", Placed: PlacedModule{Module: "app"}})
|
||||||
|
if _, err := keys.Put("g14.app", stray, 99, false); err == nil {
|
||||||
|
t.Error("a module key the node key does not name the value of was taken silently")
|
||||||
|
}
|
||||||
|
if p, _, _ := keys.Of("g14"); !SamePlacement(p, after) {
|
||||||
|
t.Errorf("a stray module key moved the placement: %+v", p)
|
||||||
|
}
|
||||||
|
// The holder's next write puts the module key right again.
|
||||||
|
repair, _ := keys.Writes("g14", &after)
|
||||||
|
if len(repair) != 2 || repair[0].Key != "g14.app" || repair[0].Base != 99 || !repair[1].Commit {
|
||||||
|
t.Fatalf("the repair writes %+v", repair)
|
||||||
|
}
|
||||||
|
apply(repair, -1)
|
||||||
|
// No placement: the legacy key first, then the node key, then the module keys; none is read after.
|
||||||
|
gone, _ := keys.Writes("g14", nil)
|
||||||
|
kinds = nil
|
||||||
|
for _, w := range gone {
|
||||||
|
kinds = append(kinds, w.Key)
|
||||||
|
}
|
||||||
|
if strings.Join(kinds, " ") != "placement.g14 g14.node g14.app g14.new" {
|
||||||
|
t.Errorf("Writes(nil) = %v", kinds)
|
||||||
|
}
|
||||||
|
apply(gone, 2)
|
||||||
|
if _, _, ok := keys.Of("g14"); ok {
|
||||||
|
t.Error("a node key deleted still places its node")
|
||||||
|
}
|
||||||
|
if _, err := DecodePlacementNode([]byte(`{"v":1,"node":"g14","modules":[{"module":"a","digest":"x"},{"module":"a","digest":"x"}]}`)); err == nil {
|
||||||
|
t.Error("a node key naming a module twice reads")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
t.Run("the catalogue's check", func(t *testing.T) {
|
t.Run("the catalogue's check", func(t *testing.T) {
|
||||||
held := func(m string) bool { return m != "ghost" }
|
held := func(m string) bool { return m != "ghost" }
|
||||||
enrolled := func(n string) bool { return n != "stranger" }
|
enrolled := func(n string) bool { return n != "stranger" }
|
||||||
|
|||||||
@@ -12,6 +12,13 @@ import (
|
|||||||
// cannot drift: until 2026-10-30 the grants part wrote its own wrapper ({node, module, digest, membership, published,
|
// cannot drift: until 2026-10-30 the grants part wrote its own wrapper ({node, module, digest, membership, published,
|
||||||
// retired}) there while the tool runner read the value as a bare membership, and served nothing (issue 774).
|
// retired}) there while the tool runner read the value as a bare membership, and served nothing (issue 774).
|
||||||
|
|
||||||
|
// GrantsBucket is the grants part's state memberships (buscompose.GrantsMembershipsBucket, which a test keeps agreeing),
|
||||||
|
// and GrantsKey an assignment's key in it, `<node>.<module>`: where a runtime reads its membership (FollowMembership).
|
||||||
|
const GrantsBucket = "mesh-grants_memberships"
|
||||||
|
|
||||||
|
// GrantsKey is an assignment's key in GrantsBucket.
|
||||||
|
func GrantsKey(node, module string) string { return node + "." + module }
|
||||||
|
|
||||||
// EncodeGrantsMembership is the value of a membership's key in the grants part's memberships: the bare membership.
|
// EncodeGrantsMembership is the value of a membership's key in the grants part's memberships: the bare membership.
|
||||||
func EncodeGrantsMembership(m Membership) []byte {
|
func EncodeGrantsMembership(m Membership) []byte {
|
||||||
raw, _ := json.Marshal(m) // a Membership holds nothing json cannot write
|
raw, _ := json.Marshal(m) // a Membership holds nothing json cannot write
|
||||||
|
|||||||
+34
-16
@@ -1,24 +1,26 @@
|
|||||||
package process
|
package process
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/nats-io/nats.go"
|
"github.com/nats-io/nats.go"
|
||||||
|
"github.com/nats-io/nats.go/jetstream"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||||
)
|
)
|
||||||
|
|
||||||
// FollowMembership reads this account's own membership and follows it (novox/hq ADR 0160): apply is called
|
// FollowMembership reads this account's own membership and follows it (novox/hq ADR 0160): apply is called
|
||||||
// with every membership the mesh issues this account's node and module, the one already issued first and
|
// with every membership the mesh issues this account's node and module, the one already issued first and
|
||||||
// each later one as it is published, so a change reaches a running process without a restart. A body that is
|
// each later one as it is issued, so a change reaches a running process without a restart. A body that is
|
||||||
// not a membership is handed to onBad (nil to ignore it) and skipped.
|
// not a membership is handed to onBad (nil to ignore it) and skipped.
|
||||||
//
|
//
|
||||||
// The subject is followed before it is read, so an issue landing between the two is applied, not missed. The
|
// It reads the membership from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7),
|
||||||
// read is the subject-addressed direct get, the one request this account may make of the stream; no
|
// watched so the value there now comes first and each change after it; and it hears the subject the grants part
|
||||||
// membership issued yet, or the read failing, is not an error: the process follows the subject and is told
|
// publishes on until the lease write. The assignments stream is no longer read: the bus module's tools delete it
|
||||||
// when one arrives. The returned error is a refusal to follow at all.
|
// at the lease write (research 072 row 33). A state not there yet, or not readable, is told to onBad and the
|
||||||
|
// subject is followed alone. The returned error is a refusal to follow at all.
|
||||||
func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(error)) error {
|
func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(error)) error {
|
||||||
if c.account.Node == "" || c.account.Module == "" {
|
if c.account.Node == "" || c.account.Module == "" {
|
||||||
return fmt.Errorf("the account names no node or module, so it has no membership")
|
return fmt.Errorf("the account names no node or module, so it has no membership")
|
||||||
@@ -29,8 +31,8 @@ func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(er
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
take := func(body []byte) {
|
take := func(body []byte) {
|
||||||
var issued membership.Membership
|
issued, err := membership.DecodeGrantsMembership(body)
|
||||||
if err := json.Unmarshal(body, &issued); err != nil {
|
if err != nil {
|
||||||
bad(fmt.Errorf("a membership arrived that is not one: %w", err))
|
bad(fmt.Errorf("a membership arrived that is not one: %w", err))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -40,14 +42,30 @@ func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(er
|
|||||||
if _, err := c.nc.Subscribe(subject, func(m *nats.Msg) { take(m.Data) }); err != nil {
|
if _, err := c.nc.Subscribe(subject, func(m *nats.Msg) { take(m.Data) }); err != nil {
|
||||||
return fmt.Errorf("cannot follow %s: %w", subject, err)
|
return fmt.Errorf("cannot follow %s: %w", subject, err)
|
||||||
}
|
}
|
||||||
got, err := c.nc.Request(membership.DirectGetSubject(c.account.Node, c.account.Module), nil, 5*time.Second)
|
key := membership.GrantsKey(c.account.Node, c.account.Module)
|
||||||
switch {
|
js, err := jetstream.New(c.nc)
|
||||||
case err != nil:
|
if err != nil {
|
||||||
bad(fmt.Errorf("cannot read the membership issued on %s yet (%v); following it", subject, err))
|
bad(fmt.Errorf("cannot read %s of %s (%v); following %s alone", key, membership.GrantsBucket, err, subject))
|
||||||
case got.Header.Get("Status") != "" || len(got.Data) == 0:
|
return nil
|
||||||
// None issued yet: the follow above delivers it.
|
|
||||||
default:
|
|
||||||
take(got.Data)
|
|
||||||
}
|
}
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
kv, err := js.KeyValue(ctx, membership.GrantsBucket)
|
||||||
|
if err != nil {
|
||||||
|
bad(fmt.Errorf("cannot read %s of %s yet (%v); following %s alone", key, membership.GrantsBucket, err, subject))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
w, err := kv.Watch(context.Background(), key)
|
||||||
|
if err != nil {
|
||||||
|
bad(fmt.Errorf("cannot watch %s of %s (%v); following %s alone", key, membership.GrantsBucket, err, subject))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
go func() {
|
||||||
|
for e := range w.Updates() {
|
||||||
|
if e != nil && e.Operation() == jetstream.KeyValuePut {
|
||||||
|
take(e.Value())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
+22
-3
@@ -43,9 +43,28 @@ func (s State) bucket() string { return s.Module + "_" + s.Name }
|
|||||||
// (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_<bucket>, so it is what withRefusal watches: a bucket
|
// (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_<bucket>, so it is what withRefusal watches: a bucket
|
||||||
// not granted is refused there, promptly, as nats.ErrPermissionViolation.
|
// not granted is refused there, promptly, as nats.ErrPermissionViolation.
|
||||||
func (c *Conn) State(name string) State {
|
func (c *Conn) State(name string) State {
|
||||||
bucket := c.account.Module + "_" + name
|
return c.bound(c.account.Module, name, func() error { return c.refusedUnannounced(name) })
|
||||||
return State{Module: c.account.Module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
|
}
|
||||||
if err := c.refusedUnannounced(name); err != nil {
|
|
||||||
|
// Read is another module's state this connection's module reads, named as its manifest's `reads` names it
|
||||||
|
// (`<module>.<state>`, novox/hq ADR 0495): bound like State to the bucket <module>_<state>, so Get, Keys and Watch
|
||||||
|
// work as far as the account's read grant goes, and a write is refused by the bus (issue 774: route-proxy follows
|
||||||
|
// the seat mesh-provisioner's plans). A name that is not `<module>.<state>` fails at the call that needed it.
|
||||||
|
func (c *Conn) Read(read string) State {
|
||||||
|
module, name, ok := strings.Cut(read, ".")
|
||||||
|
return c.bound(module, name, func() error {
|
||||||
|
if !ok || module == "" || name == "" || strings.Contains(name, ".") {
|
||||||
|
return fmt.Errorf("%q is not a state another module keeps, <module>.<state>", read)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// bound is module's state name on this connection, checked by refused before every bind.
|
||||||
|
func (c *Conn) bound(module, name string, refused func() error) State {
|
||||||
|
bucket := module + "_" + name
|
||||||
|
return State{Module: module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
|
||||||
|
if err := refused(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
js, err := jetstream.New(c.nc)
|
js, err := jetstream.New(c.nc)
|
||||||
|
|||||||
@@ -353,6 +353,18 @@ func TestStateWatchRefusesAnUngrantedBucketNamingIt(t *testing.T) {
|
|||||||
if err == nil || !errors.Is(err, nats.ErrPermissionViolation) || !strings.Contains(err.Error(), "mesh-mail_unrelated") {
|
if err == nil || !errors.Is(err, nats.ErrPermissionViolation) || !strings.Contains(err.Error(), "mesh-mail_unrelated") {
|
||||||
t.Fatalf("Watch on an ungranted bucket: %v, want a refusal naming mesh-mail_unrelated", err)
|
t.Fatalf("Watch on an ungranted bucket: %v, want a refusal naming mesh-mail_unrelated", err)
|
||||||
}
|
}
|
||||||
|
// Another module's state, read as `reads` names it (issue 774): bound to its bucket, refused naming it when no read
|
||||||
|
// is granted; a name that is not <module>.<state> is refused before the bus is asked.
|
||||||
|
if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-provisioner_plans"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
err = conn.Read("mesh-provisioner.plans").Watch("", func(stdio.StateChange) error { return nil })
|
||||||
|
if err == nil || !errors.Is(err, nats.ErrPermissionViolation) || !strings.Contains(err.Error(), "mesh-provisioner_plans") {
|
||||||
|
t.Fatalf("Read of an ungranted state: %v, want a refusal naming mesh-provisioner_plans", err)
|
||||||
|
}
|
||||||
|
if _, err := conn.Read("plans").Get("x"); err == nil || !strings.Contains(err.Error(), "<module>.<state>") {
|
||||||
|
t.Fatalf("Read of a name that is no other module's state: %v", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 1840/S1: given a process connection granted its bucket, when it deletes a key, then the key is gone and
|
// 1840/S1: given a process connection granted its bucket, when it deletes a key, then the key is gone and
|
||||||
|
|||||||
Reference in New Issue
Block a user