jochen
e11caecdad
The controller's machine moves it from the container to a process by starting the process first and removing the container once the process is up (mesh-host's `replaces`). For that moment two controllers share the store and the bus. Checked what each does: - the seat's verbs: a queue group per seat, each call answered once. Safe. - the controller's consumers on CONTROL and EVENTS: push consumers with no delivery group, so the second bind is refused with "consumer is already bound" and serve exited. The process would restart for ever, the host would never see it up, and the container would never go. The second controller now stands by and binds when the first lets go (tested on a real bus; fails without the change). - plans: read, changed and saved whole by the 30s timer, by build outcomes, by a merge and by `plans stop`. Two timers would each ask a tier the other had just asked. Working the plans now takes a session-level advisory lock on the inventory: the timer skips while another holds it, the other paths wait for it. Build asks happen only inside plan work and are covered by the same lock.
70 lines
2.3 KiB
Go
70 lines
2.3 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
)
|
|
|
|
// novox/hq issue 213: while a machine hands its controller over, the new controller (the process)
|
|
// is started while the old one (the container) still holds the controller's consumers. It must not
|
|
// exit — the host would read that as a replacement that did not come up and never remove the
|
|
// container — and must not act on what the old one is handed. It stands by, and takes the consumers
|
|
// when the old one lets go.
|
|
func TestNatsASecondControllerStandsByAndTakesOverWhenTheFirstLetsGo(t *testing.T) {
|
|
js := aBus(t)
|
|
was := StandbyPoll
|
|
StandbyPoll = 50 * time.Millisecond
|
|
defer func() { StandbyPoll = was }()
|
|
|
|
old := &counted{}
|
|
_, stopOld := servingOn(t, js, old)
|
|
eventually(t, "the first controller binding its consumer", func() bool {
|
|
info, err := js.Context().ConsumerInfo("CONTROL", broker.ControllerName)
|
|
return err == nil && info.PushBound
|
|
})
|
|
|
|
// The new one, on a connection of its own as the process would have.
|
|
second, err := broker.Dial(os.Getenv("MESH_TEST_NATS"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(second.Close)
|
|
fresh := &counted{}
|
|
s := &Server{inbound: Nats(second), bus: OverNATS{Conn: second.Conn(), JS: second.Context()},
|
|
listener: fresh, log: quiet()}
|
|
ctx, stopNew := context.WithCancel(context.Background())
|
|
defer stopNew()
|
|
ended := make(chan error, 1)
|
|
go func() { ended <- s.Serve(ctx) }()
|
|
|
|
select {
|
|
case err := <-ended:
|
|
t.Fatalf("the second controller stopped instead of standing by: %v", err)
|
|
case <-time.After(500 * time.Millisecond):
|
|
}
|
|
|
|
report := func(declared string) {
|
|
body, _ := json.Marshal(Report{Node: "anchor", Declared: declared, Applied: []string{"store"}})
|
|
if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
report("d1")
|
|
eventually(t, "the holding controller hearing the report", func() bool { return old.count() == 1 })
|
|
if fresh.count() != 0 {
|
|
t.Fatal("the controller standing by acted on a report the holder was handed")
|
|
}
|
|
|
|
stopOld()
|
|
report("d2")
|
|
eventually(t, "the second controller taking over once the first let go", func() bool { return fresh.count() == 1 })
|
|
if old.count() != 1 {
|
|
t.Errorf("the first controller heard %d reports", old.count())
|
|
}
|
|
}
|