diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 00ce9a28..e331c3a4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -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() { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index ba23e94e..6e324839 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index 3155a771..5563451a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -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" {