diff --git a/cmd/mesh-controller/plan_looks.go b/cmd/mesh-controller/plan_looks.go index 6e4f155d..1bfdbf7d 100644 --- a/cmd/mesh-controller/plan_looks.go +++ b/cmd/mesh-controller/plan_looks.go @@ -171,6 +171,9 @@ func startRolloutEvents(ctx context.Context, open *stores, js *broker.JetStream) if err := startProvAppliedLooks(ctx, js); err != nil { fmt.Printf("plans: the provisioner clients' words could not be watched (%v); a walk waits for them until the next look\n", err) } + if err := startPlanMovedLooks(ctx, js); err != nil { + fmt.Printf("plans: the plans moved by other processes could not be followed (%v)\n", err) + } startPlanLooks(ctx, open) } @@ -507,6 +510,36 @@ func startProvAppliedLooks(ctx context.Context, js *broker.JetStream) error { return nil } +// startPlanMovedLooks has the plans looked at when any process says one moved: the serving controller follows its own +// `plan-moved`, so a plan a person moved with a verb at the control node's terminal — a process this one did not spawn, +// whose saves it does not hear — is looked at all the same (novox/hq ADR 0406 §2, §5, issue 974). New messages only; the +// first look reads the rest. A look its own save asks is the same look: asks are one. +func startPlanMovedLooks(ctx context.Context, js *broker.JetStream) error { + api, err := jetstream.New(js.Conn()) + if err != nil { + return err + } + stream, err := api.Stream(ctx, broker.EventsStream) + if err != nil { + return fmt.Errorf("the events stream: %w", err) + } + consumer, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{link.SeatEventSubject(link.MeshControllerSeat, link.KeyPlanMoved)}, + DeliverPolicy: jetstream.DeliverNewPolicy}) + if err != nil { + return fmt.Errorf("following the plans moved: %w", err) + } + consuming, err := consumer.Consume(func(jetstream.Msg) { askToLook() }) + if err != nil { + return fmt.Errorf("following the plans moved: %w", err) + } + go func() { + <-ctx.Done() + consuming.Stop() + }() + return nil +} + // --- the end of a walk whose end is not known --- var walkEnds = struct { diff --git a/cmd/mesh-controller/plan_looks_test.go b/cmd/mesh-controller/plan_looks_test.go index 3f4544eb..c2076a4b 100644 --- a/cmd/mesh-controller/plan_looks_test.go +++ b/cmd/mesh-controller/plan_looks_test.go @@ -9,6 +9,7 @@ import ( "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" "github.com/novox/mesh-controller/internal/testbus" ) @@ -315,3 +316,36 @@ func TestKeysThatCannotBeReadAreWatchedByTimersAndTheLookFails(t *testing.T) { t.Error("no timer of its own watches the wait") } } + +// A plan moved by a process the serving controller did not spawn — a verb run at the control node's terminal — is +// looked at: the controller follows its own plan-moved (issue 974, round 1). +func TestAPlanMovedByAnotherProcessAsksForALook(t *testing.T) { + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if err := broker.AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + var failures, looks atomic.Int32 + l := lookerCounting(&failures, &looks, nil) + lookerMu.Lock() + theLooker = l + lookerMu.Unlock() + t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock() }) + ctx, stop := context.WithCancel(context.Background()) + defer stop() + go l.run(ctx) + if err := startPlanMovedLooks(ctx, js); err != nil { + t.Fatal(err) + } + legacy, err := js.Conn().JetStream() + if err != nil { + t.Fatal(err) + } + if _, err := legacy.Publish(link.SeatEventSubject(link.MeshControllerSeat, link.KeyPlanMoved), []byte(`{}`)); err != nil { + t.Fatal(err) + } + eventually(t, "a look on the plan moved", func() bool { return looks.Load() >= 1 }) +}