mesh-tools
97a2d841c7
The controller plans config, templates and moved providers' provisions, hands bus accounts to the vault and judges the new manifest words, on the feature's SDK and node-engine validator (feature 57, config and provisioning).
312 lines
11 KiB
Go
312 lines
11 KiB
Go
package link
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
)
|
|
|
|
// The calls bucket against a real server (novox/hq to-be 45 §6): whether a call's outcome is
|
|
// readable by id from a controller that did not serve it is a claim about what the bus keeps.
|
|
|
|
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
|
|
}
|
|
|
|
// waitKept waits until the keeper holds a call in a state, the writes being queued.
|
|
func waitKept(t *testing.T, keeper CallKeeper, id, state string) Call {
|
|
t.Helper()
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for {
|
|
c, found, err := keeper.Kept(context.Background(), id)
|
|
if err == nil && found && c.State == state {
|
|
return c
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("%s never kept as %q: %+v %v %v", id, state, c, found, err)
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// **A controller restart keeps every call's outcome** (to-be 45 Phase 0): a call one controller
|
|
// answered is read whole by id from the next, and a call it left running is said abandoned rather
|
|
// than running for ever.
|
|
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)
|
|
}
|
|
}
|
|
|
|
// racingRecent wraps a CallKeeper so a finish lands right after its Recent() snapshot is taken, as
|
|
// if that finish raced the moment the snapshot was read (novox/hq issue 711): the snapshot still
|
|
// says a call is running, but by the time anything acts on that entry the record has already moved
|
|
// on.
|
|
type racingRecent struct {
|
|
CallKeeper
|
|
thenLand func()
|
|
}
|
|
|
|
func (r racingRecent) Recent(ctx context.Context) ([]Call, error) {
|
|
out, err := r.CallKeeper.Recent(ctx)
|
|
if err == nil && r.thenLand != nil {
|
|
r.thenLand()
|
|
}
|
|
return out, err
|
|
}
|
|
|
|
// **A finish recorded before the sweep reaches a call stands, with its kept answer** (novox/hq issue
|
|
// 711): a stale "running" snapshot is no excuse to overwrite a record that has since said how the
|
|
// call ended. A caller reading `abandoned` for a call that succeeded would redo work already done.
|
|
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())
|
|
}
|
|
}
|
|
|
|
// A call left running with no finish recorded is still marked abandoned — unchanged (novox/hq issue
|
|
// 711): only a finish actually recorded for it may stand instead.
|
|
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())
|
|
}
|
|
}
|
|
|
|
// delayedKV wraps a jetstream.KeyValue so a hook runs just before Update, to land a write between a
|
|
// reader's Get and its own compare-and-set write.
|
|
type delayedKV struct {
|
|
jetstream.KeyValue
|
|
before func()
|
|
}
|
|
|
|
func (d delayedKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) {
|
|
d.before()
|
|
return d.KeyValue.Update(ctx, key, value, revision)
|
|
}
|
|
|
|
// **The race where a finish lands after the sweep's own read but before its write: the finish
|
|
// wins** (novox/hq issue 711). The sweep's write is a compare-and-set on the revision it read, so a
|
|
// finish landing in between invalidates it; the sweep rereads, finds the finish, and leaves it alone.
|
|
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())
|
|
}
|
|
}
|
|
|
|
// alwaysMismatchKV wraps a jetstream.KeyValue whose Update always fails with a revision mismatch, as
|
|
// if some other writer kept winning the race forever (novox/hq issue 711 review round 1).
|
|
type alwaysMismatchKV struct {
|
|
jetstream.KeyValue
|
|
tries *int
|
|
}
|
|
|
|
func (a alwaysMismatchKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) {
|
|
*a.tries++
|
|
return 0, jetstream.ErrKeyRevisionMismatch
|
|
}
|
|
|
|
// **MarkAbandoned gives up after markAbandonedTries rather than loop forever** (novox/hq issue 711
|
|
// review round 1): a record whose compare-and-set keeps losing is left as it stands, said in an
|
|
// error naming the call, not retried without end.
|
|
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)
|
|
}
|
|
}
|
|
|
|
// The bucket holds the last thousand calls or fourteen days: a call is two keys, so the stream under
|
|
// it holds twice as many messages, and asserting it again keeps the count.
|
|
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)
|
|
}
|
|
}
|
|
|
|
// An answer larger than the bound is cut and says so, and the record is still written.
|
|
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)
|
|
}
|
|
}
|
|
|
|
// Who asked is read from the inbox the answer goes to.
|
|
func TestTheCallerIsTheInboxsPrincipal(t *testing.T) {
|
|
for reply, want := range map[string]string{
|
|
"_INBOX.node-tools.g14.Pe3sGzAtv8jUBYQ6sKSvKz.1": "node-tools.g14",
|
|
"_INBOX.jochen.Pe3sGzAtv8jUBYQ6sKSvKz": "jochen",
|
|
"_INBOX.x.1": "",
|
|
"mesh.control.one": "",
|
|
"": "",
|
|
} {
|
|
if got := callerOf(reply); got != want {
|
|
t.Errorf("%q: %q, want %q", reply, got, want)
|
|
}
|
|
}
|
|
}
|