Assigner: claim the seat mesh-placement with its stream, invokes and the resolved values it composes with (ADR 0567 §1, §6, §8, §9, issue 1126)
This commit is contained in:
@@ -23,6 +23,7 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
@@ -50,6 +51,7 @@ func main() {
|
||||
// readBuckets are the buckets of other parts the assigner reads, each named in its manifest's reads (the manifest test
|
||||
// holds the two lists equal), so the controller's broker grants exactly these.
|
||||
var readBuckets = []string{
|
||||
buscompose.ResolvedBucket,
|
||||
provider.AssignmentsBucket,
|
||||
delivery.BuildsBucket,
|
||||
delivery.ControllerGrantsBucket,
|
||||
@@ -160,6 +162,7 @@ func (e emitting) Emit(eventType string, body any) error { return e.Conn.Emit(ev
|
||||
// watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and
|
||||
// watched only for keys, so a manifest that arrives wakes a composition that waits for it.
|
||||
var watched = []string{
|
||||
buscompose.ResolvedBucket,
|
||||
provider.AssignmentsBucket,
|
||||
delivery.ControllerGrantsBucket,
|
||||
delivery.ControllerComposedBucket,
|
||||
|
||||
@@ -18,8 +18,8 @@ import (
|
||||
|
||||
// The assigner's tools (novox/hq ADR 0546), served through the tool runner from its bundle: a process never serves (ADR
|
||||
// 0436, mesh-sdk process). They write and read the puts; the assigner's process takes each put further (puts.go) and says
|
||||
// every step as the event put. They are the module's own tools, not a seat's verbs: a new seat needs a change of the
|
||||
// controller, which takes none; they become the verbs of a seat once composing leaves the controller.
|
||||
// every step as the event put. They are the module's own tools; the seat mesh-placement's verbs are served beside them
|
||||
// (seat_tools.go, ADR 0567 §10).
|
||||
|
||||
// putsState is the assigner's state of puts, as the runtime names a module's own state.
|
||||
const putsState = "puts"
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||
@@ -358,7 +359,25 @@ func (r worldReader) Manifest(ctx context.Context, digest string) ([]byte, error
|
||||
return []byte(m.JSON), nil
|
||||
}
|
||||
|
||||
func (worldReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil }
|
||||
// Values is one module's config values on one node as mesh-config resolved them, each a value its source lays
|
||||
// (catalogue.LaidValue): the holder composes every node a proposal touches with them (ADR 0567 §9).
|
||||
func (r worldReader) Values(_ context.Context, node, module string) (map[string]any, error) {
|
||||
var out map[string]any
|
||||
prefix := node + "." + module + "."
|
||||
for key, raw := range r.w.read(buscompose.ResolvedBucket) {
|
||||
name, ok := strings.CutPrefix(key, prefix)
|
||||
if !ok || strings.Contains(name, ".") {
|
||||
continue
|
||||
}
|
||||
if v, laid := catalogue.LaidValue(raw); laid {
|
||||
if out == nil {
|
||||
out = map[string]any{}
|
||||
}
|
||||
out[name] = v
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (worldReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
|
||||
return nil, nil
|
||||
@@ -368,8 +387,8 @@ func (worldReader) Credentials(context.Context, string, []compose.CredentialKey,
|
||||
return nil, nil, nil
|
||||
}
|
||||
|
||||
// Build, Archive and Served are read only for a node composing itself (frombus.Scope without Every): the check of every
|
||||
// node composes from the manifests the indexer read, reads no build and no archive, and reads no config value.
|
||||
// Build and Archive are read only for a node composing itself (frombus.Scope without Every): the check of every node
|
||||
// composes from the manifests the indexer read, and reads no build and no archive.
|
||||
func (worldReader) Build(context.Context, string, string) (*delivery.BuilderBuild, error) {
|
||||
return nil, nil
|
||||
}
|
||||
@@ -378,8 +397,18 @@ func (worldReader) Archive(_ context.Context, url, _ string) ([]byte, error) {
|
||||
return nil, fmt.Errorf("the composition check reads no archive (%s)", url)
|
||||
}
|
||||
|
||||
func (worldReader) Served(context.Context, string, string, []string) (map[string]any, error) {
|
||||
return nil, nil
|
||||
func (r worldReader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) {
|
||||
all, err := r.Values(ctx, node, module)
|
||||
var out map[string]any
|
||||
for _, k := range keys {
|
||||
if v, ok := all[k]; ok {
|
||||
if out == nil {
|
||||
out = map[string]any{}
|
||||
}
|
||||
out[k] = v
|
||||
}
|
||||
}
|
||||
return out, err
|
||||
}
|
||||
|
||||
// change is a candidate's change for the composition check, or why the composition waits for it.
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
func load(t *testing.T) catalogue.Manifest {
|
||||
@@ -32,7 +33,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-build-go.steps", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants",
|
||||
want := []string{"mesh-build-go.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants",
|
||||
"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
@@ -58,12 +59,27 @@ func TestManifest(t *testing.T) {
|
||||
t.Errorf("uses = %v, want exactly node-build-go and operator-channel", m.Uses)
|
||||
}
|
||||
// ADR 0546: it serves its own tools (put, release, puts) from its bundle through the tool runner (the artifact loads the
|
||||
// module), claims no seat (a new seat needs a change of the controller), says each put's step as the event put, and
|
||||
// calls the controller's warranted and the bus node's backup holder.
|
||||
if len(m.Claims) != 0 || !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts"}) {
|
||||
t.Errorf("the assigner claims %v and serves tools %v, want no claim and its three put tools", m.Claims, m.Tools)
|
||||
// module), says each put's step as the event put, and calls the controller's warranted and the bus node's backup holder.
|
||||
// ADR 0567 §1, §6-§10: it holds the seat mesh-placement, serving its seven verbs; keeps the seat's events on its stream
|
||||
// placement for 30 days; calls the controller's export, its four verbs that keep its assignments in step and its
|
||||
// hand-act log; and reads mesh-config's resolved values to compose every node a proposal touches.
|
||||
if len(m.Claims) != 1 || m.Claims[0].Name != catalogue.PlacementSeat || m.Claims[0].At() != catalogue.ScopeMesh ||
|
||||
!slices.Equal(m.Claims[0].Serves, []string{"placement", "proposals", "propose", "withdraw", "again", "adopt", "hand-back"}) {
|
||||
t.Errorf("the assigner claims %+v, want mesh-placement with its seven verbs", m.Claims)
|
||||
}
|
||||
if want := []string{"seat:mesh-controller.warranted", "seat:node-backup.backed-up", "seat:node-backup.now"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) {
|
||||
if seat, _ := catalogue.SeatNamed(catalogue.PlacementSeat); catalogue.CanHold(m, seat) != nil {
|
||||
t.Errorf("the assigner cannot hold the seat: %v", catalogue.CanHold(m, seat))
|
||||
}
|
||||
if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts"}) {
|
||||
t.Errorf("the assigner serves tools %v, want its three put tools", m.Tools)
|
||||
}
|
||||
if len(m.Streams) != 1 || m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays ||
|
||||
!slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(delivery.PlacementEvents))) {
|
||||
t.Errorf("streams = %+v, want placement carrying the seat's four events for %d days", m.Streams, delivery.PlacementStreamDays)
|
||||
}
|
||||
if want := []string{"seat:mesh-controller.assign", "seat:mesh-controller.hand-act", "seat:mesh-controller.pin",
|
||||
"seat:mesh-controller.placement-export", "seat:mesh-controller.unassign", "seat:mesh-controller.unpin",
|
||||
"seat:mesh-controller.warranted", "seat:node-backup.backed-up", "seat:node-backup.now"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
if !slices.Equal(m.Emits, []string{"put"}) {
|
||||
|
||||
@@ -2,8 +2,24 @@
|
||||
"module": "mesh-delivery-assigner",
|
||||
"version": "1",
|
||||
"slug": "assigner",
|
||||
"claims": [
|
||||
{
|
||||
"name": "mesh-placement",
|
||||
"scope": "mesh",
|
||||
"serves": [
|
||||
"placement",
|
||||
"proposals",
|
||||
"propose",
|
||||
"withdraw",
|
||||
"again",
|
||||
"adopt",
|
||||
"hand-back"
|
||||
]
|
||||
}
|
||||
],
|
||||
"reads": [
|
||||
"mesh-build-go.steps",
|
||||
"mesh-config.resolved",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-controller.builds",
|
||||
"mesh-controller.composed",
|
||||
@@ -21,9 +37,15 @@
|
||||
"operator-channel"
|
||||
],
|
||||
"invokes": [
|
||||
"seat:mesh-controller.assign",
|
||||
"seat:mesh-controller.hand-act",
|
||||
"seat:mesh-controller.pin",
|
||||
"seat:mesh-controller.placement-export",
|
||||
"seat:mesh-controller.unassign",
|
||||
"seat:mesh-controller.unpin",
|
||||
"seat:mesh-controller.warranted",
|
||||
"seat:node-backup.now",
|
||||
"seat:node-backup.backed-up"
|
||||
"seat:node-backup.backed-up",
|
||||
"seat:node-backup.now"
|
||||
],
|
||||
"emits": [
|
||||
"put"
|
||||
@@ -43,6 +65,18 @@
|
||||
{ "name": "puts", "history": 20 },
|
||||
{ "name": "wants", "history": 20 }
|
||||
],
|
||||
"streams": [
|
||||
{
|
||||
"name": "placement",
|
||||
"events": [
|
||||
"placement-proposed",
|
||||
"placement-judged",
|
||||
"placement-changed",
|
||||
"placement-withdrawn"
|
||||
],
|
||||
"max-age-days": 30
|
||||
}
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user