issue 974: the serving controller follows its own plan-moved, so a plan moved from the terminal is looked at
This commit is contained in:
@@ -171,6 +171,9 @@ func startRolloutEvents(ctx context.Context, open *stores, js *broker.JetStream)
|
|||||||
if err := startProvAppliedLooks(ctx, js); err != nil {
|
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)
|
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)
|
startPlanLooks(ctx, open)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -507,6 +510,36 @@ func startProvAppliedLooks(ctx context.Context, js *broker.JetStream) error {
|
|||||||
return nil
|
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 ---
|
// --- the end of a walk whose end is not known ---
|
||||||
|
|
||||||
var walkEnds = struct {
|
var walkEnds = struct {
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
|
|
||||||
"github.com/novox/mesh-controller/internal/broker"
|
"github.com/novox/mesh-controller/internal/broker"
|
||||||
"github.com/novox/mesh-controller/internal/inventory"
|
"github.com/novox/mesh-controller/internal/inventory"
|
||||||
|
"github.com/novox/mesh-controller/internal/link"
|
||||||
"github.com/novox/mesh-controller/internal/testbus"
|
"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")
|
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 })
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user