Assigner: the outside wait's settings are read from the program's environment, and the bed has a keeper

Builds: design 47 G §706, §633

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-11 22:53:42 +01:00
parent 00ec7af34e
commit 6cec46cc77
3 changed files with 26 additions and 6 deletions
@@ -204,6 +204,26 @@ func startBed(t *testing.T, leaveOut string) *bed {
} else {
b.buckets[marksBucket] = kv
}
// The condition keeper's verb conditions (design 47 G §706): no condition is open, so a failing check is the delivery's after the
// settle, which the bed sets to a second.
for _, subject := range []string{"mesh.seat.mesh-conditions.tool.conditions", "mesh.seat.mesh-conditions.call.conditions.*", "mesh.seat.mesh-conditions.tool.conditions.*"} {
sub, err := nc.Subscribe(subject, func(m *nats.Msg) { _ = m.Respond([]byte(`{"result":{"conditions":[],"open":0},"node":"n0"}`)) })
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
subscribed(t, nc)
// The condition keeper's verb conditions (design 47 G §706): no condition is open, so a failing check is the delivery's after the
// settle, which the bed sets to a second.
for _, subject := range []string{"mesh.seat.mesh-conditions.tool.conditions", "mesh.seat.mesh-conditions.call.conditions.*", "mesh.seat.mesh-conditions.tool.conditions.*"} {
sub, err := nc.Subscribe(subject, func(m *nats.Msg) { _ = m.Respond([]byte(`{"result":{"conditions":[],"open":0},"node":"n0"}`)) })
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
subscribed(t, nc)
// The Go build seat's work queue, as the broker derives it from the seat's accepted kinds (delivery.AskStream).
seat, _ := buildstep.AskStream(buildstep.AskGo)
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: seat, Subjects: []string{"mesh.seat.node-build-go.accept.>"},
@@ -230,7 +250,7 @@ func (b *bed) startStoppable(t *testing.T) (<-chan error, func()) {
if err := os.WriteFile(broker, cred, 0o600); err != nil {
t.Fatal(err)
}
env := map[string]string{"MESH_BROKER_FILE": broker}
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_ASSIGNER_OUTSIDE_SETTLE_SECONDS": "1"}
ctx, stop := context.WithCancel(b.ctx)
done := make(chan error, 1)
go func() {
@@ -120,7 +120,7 @@ var keptStates = []announce.State{
{Name: copiesState, Readable: true},
{Name: writtenState, Readable: true},
{Name: buckets.WaitsState, Readable: true},
{Name: buckets.InputsState}, // read by the tool process alone: the verb input and assigner_inputs (design 47 G §354)
{Name: buckets.InputsState, Readable: true}, // read by the tool process alone: the verb input and assigner_inputs (design 47 G §354)
}
// kvSet is every bucket the assigner opened, by name.
@@ -266,7 +266,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}()
iv := &catalogViews{js: js, account: account, call: said.Call, writer: wireOf(writer), logf: logger.Printf,
status: process.AwaitBucket(ctx, js, conn, firstViewsBucket, "assigner_first_views reads nothing and assigner_first_view_again is not heard", logger.Printf)}
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, logger)
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, logger, getenv)
}
// findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the
@@ -348,7 +348,7 @@ var watched = []string{
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs kvSet, say func(oc.Condition) error, findings assign.Findings, bus caller, iv *catalogViews, logger *log.Logger) error {
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs kvSet, say func(oc.Condition) error, findings assign.Findings, bus caller, iv *catalogViews, logger *log.Logger, getenv func(string) string) error {
w := newWorld()
dirty := make(chan struct{}, 1)
wake := func() {
@@ -525,7 +525,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
// The wait on an outside cause (outside.go, design 47 G §706): the keeper's three events say when to judge again, and the
// seat mesh-wants' wait-held and wait-ended are said; registered before the hub starts.
if bus != nil && kvs[buckets.Waits] != nil {
bound, settle := outsideSettings(os.Getenv)
bound, settle := outsideSettings(getenv)
p.outside = newOutsideActs(bus.CallKept, wake, logger.Printf, bound, settle)
for _, subject := range []string{mcseat.SubjectConditionRaised, mcseat.SubjectConditionChanged, mcseat.SubjectConditionCleared} {
subject := subject
@@ -124,7 +124,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
}
kvs[buckets.Puts].(*watchKV).closed = true
kvs[buckets.Wants].(*watchKV).wrote = wrote
go follow(ctx, nil, nil, kvs, func(oc.Condition) error { return nil }, assign.Findings{}, nil, nil, log.New(io.Discard, "", 0))
go follow(ctx, nil, nil, kvs, func(oc.Condition) error { return nil }, assign.Findings{}, nil, nil, log.New(io.Discard, "", 0), func(string) string { return "" })
select {
case key := <-wrote:
if key != "want.ace.node" {