//go:build integration package link import ( "bytes" "context" "encoding/json" "log" "strings" "testing" "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" ) func callsBucket(t *testing.T) *BusCalls { t.Helper() js := aBus(t) api, err := jetstream.New(js.Conn()) if err != nil { t.Fatal(err) } _ = api.DeleteKeyValue(t.Context(), broker.CallsBucket) if err := js.EnsureControllerBuckets(); err != nil { t.Fatal(err) } keeper, err := CallsOnTheBus(t.Context(), js.Conn()) if err != nil { t.Fatal(err) } return keeper } func TestNatsACallsOutcomeOutlivesItsController(t *testing.T) { keeper := callsBucket(t) first := NewCallLog() if err := first.Durably(t.Context(), keeper, "controller@one", nil); err != nil { t.Fatal(err) } a := newAnswers(t) first.serveCall("mesh-controller", "push", json.RawMessage(`{"node":"anchor","values":"secret"}`), "_INBOX.node-tools.g14.Pe3sGzAtv8jUBYQ6sKSvKz.1", func(context.Context, json.RawMessage) (any, error) { return "anchor told", nil }, a.respond, nil) recent, _ := first.Recent() finished := waitKept(t, keeper, recent[0].ID, CallAnswered) // A second call, still running when its controller stops. left := first.begin("mesh-controller", "plans", json.RawMessage(`{}`), "") waitKept(t, keeper, left.ID, CallRunning) var said bytes.Buffer second := NewCallLog() if err := second.Durably(t.Context(), keeper, "controller@two", log.New(&said, "", 0)); err != nil { t.Fatal(err) } c, found, err := second.Get(finished.ID) if err != nil || !found { t.Fatalf("the next controller cannot read %s: %v %v", finished.ID, found, err) } if !strings.Contains(string(c.Answer), "anchor told") || c.Caller != "node-tools.g14" || c.Holder != "controller@one" { t.Fatalf("kept %+v", c) } if strings.Contains(string(c.Args), "secret") { t.Fatalf("a setting was kept: %s", c.Args) } abandoned, _, _ := second.Get(left.ID) if abandoned.State != CallAbandoned || !strings.Contains(said.String(), left.ID) { t.Fatalf("a call left running reads %q, and was said: %q", abandoned.State, said.String()) } listed, err := second.Recent() if err != nil || len(listed) != 2 { t.Fatalf("the next controller lists %d calls: %v", len(listed), err) } // Its own running calls are not its predecessor's: a controller starting again under the same // name leaves them alone. mine := second.begin("mesh-controller", "status", nil, "") waitKept(t, keeper, mine.ID, CallRunning) if err := second.Durably(t.Context(), keeper, "controller@two", nil); err != nil { t.Fatal(err) } if c, _, _ := keeper.Kept(t.Context(), mine.ID); c.State != CallRunning { t.Fatalf("a controller marked its own running call %q", c.State) } } func TestNatsAFinishRacingTheSweepsSnapshotReadsFinished(t *testing.T) { keeper := callsBucket(t) running := Call{ID: "call-711-1", Seat: "mesh-controller", Verb: "push", State: CallRunning, Holder: "controller@one"} if err := keeper.Keep(t.Context(), running); err != nil { t.Fatal(err) } waitKept(t, keeper, running.ID, CallRunning) finished := running finished.State = CallAnswered finished.Answer = json.RawMessage(`"anchor told"`) wrapped := racingRecent{CallKeeper: keeper, thenLand: func() { if err := keeper.Keep(context.Background(), finished); err != nil { t.Error(err) } waitKept(t, keeper, running.ID, CallAnswered) }} var said bytes.Buffer second := NewCallLog() if err := second.Durably(t.Context(), wrapped, "controller@two", log.New(&said, "", 0)); err != nil { t.Fatal(err) } got, found, err := second.Get(running.ID) if err != nil || !found || got.State != CallAnswered || !strings.Contains(string(got.Answer), "anchor told") { t.Fatalf("its finish did not stand: %+v %v %v", got, found, err) } if strings.Contains(said.String(), running.ID) { t.Errorf("a finished call was said abandoned: %q", said.String()) } } func TestNatsARunningCallWithNoFinishIsStillAbandoned(t *testing.T) { keeper := callsBucket(t) running := Call{ID: "call-711-2", Seat: "mesh-controller", Verb: "plans", State: CallRunning, Holder: "controller@one"} if err := keeper.Keep(t.Context(), running); err != nil { t.Fatal(err) } waitKept(t, keeper, running.ID, CallRunning) var said bytes.Buffer second := NewCallLog() if err := second.Durably(t.Context(), keeper, "controller@two", log.New(&said, "", 0)); err != nil { t.Fatal(err) } got, found, err := second.Get(running.ID) if err != nil || !found || got.State != CallAbandoned || !strings.Contains(said.String(), running.ID) { t.Fatalf("got %+v %v %v; said %q", got, found, err, said.String()) } } func TestNatsAFinishRacingTheSweepsWriteWins(t *testing.T) { keeper := callsBucket(t) running := Call{ID: "call-711-3", Seat: "mesh-controller", Verb: "push", State: CallRunning, Holder: "controller@one"} if err := keeper.Keep(t.Context(), running); err != nil { t.Fatal(err) } waitKept(t, keeper, running.ID, CallRunning) finished := running finished.State = CallFinishedAfter finished.Answer = json.RawMessage(`"anchor told"`) landed := false racey := &BusCalls{kv: delayedKV{KeyValue: keeper.kv, before: func() { if landed { return } landed = true if err := keeper.Keep(context.Background(), finished); err != nil { t.Error(err) } waitKept(t, keeper, running.ID, CallFinishedAfter) }}} var said bytes.Buffer second := NewCallLog() if err := second.Durably(t.Context(), racey, "controller@two", log.New(&said, "", 0)); err != nil { t.Fatal(err) } got, found, err := second.Get(running.ID) if err != nil || !found || got.State != CallFinishedAfter || !strings.Contains(string(got.Answer), "anchor told") { t.Fatalf("the finish did not win the race: %+v %v %v", got, found, err) } if strings.Contains(said.String(), running.ID) { t.Errorf("a finished call was said abandoned: %q", said.String()) } } func TestNatsAMarkAbandonedGivesUpAfterTenTries(t *testing.T) { keeper := callsBucket(t) running := Call{ID: "call-711-4", Seat: "mesh-controller", Verb: "push", State: CallRunning, Holder: "controller@one"} if err := keeper.Keep(t.Context(), running); err != nil { t.Fatal(err) } waitKept(t, keeper, running.ID, CallRunning) tries := 0 stuck := &BusCalls{kv: alwaysMismatchKV{KeyValue: keeper.kv, tries: &tries}} _, err := stuck.MarkAbandoned(t.Context(), running.ID) if err == nil || !strings.Contains(err.Error(), running.ID) { t.Fatalf("expected an error naming %s, got %v", running.ID, err) } if tries != markAbandonedTries { t.Fatalf("tried %d times, want %d", tries, markAbandonedTries) } } func TestNatsTheCallsBucketIsBounded(t *testing.T) { callsBucket(t) js := aBus(t) if err := js.EnsureControllerBuckets(); err != nil { t.Fatal(err) } info, err := js.Context().StreamInfo("KV_" + broker.CallsBucket) if err != nil { t.Fatal(err) } if info.Config.MaxMsgs != 2*broker.KeptCallsDurably || info.Config.MaxAge != broker.CallsKeptFor { t.Fatalf("kept %d messages for %s", info.Config.MaxMsgs, info.Config.MaxAge) } } func TestNatsAnAnswerLargerThanTheBoundIsCut(t *testing.T) { keeper := callsBucket(t) big, _ := json.Marshal(map[string]string{"result": strings.Repeat("x", broker.CallAnswerBytes+10)}) c := Call{ID: "call-1-1", Seat: "mesh-controller", Verb: "plan", State: CallAnswered, Answer: big} if err := keeper.Keep(t.Context(), c); err != nil { t.Fatal(err) } got, found, err := keeper.Kept(t.Context(), "call-1-1") if err != nil || !found || !strings.Contains(string(got.Answer), "the first") { t.Fatalf("%v %v %.200s", found, err, got.Answer) } }