Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575package spindle
import ( "context" "errors" "fmt" "slices" "strings" "testing" "time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled" "tangled.org/core/jetstream" "tangled.org/core/rbac" "tangled.org/core/spindle/knotclient" "tangled.org/core/spindle/models")
var ( opCreate = jetstream.CommitOperationCreate opDelete = jetstream.CommitOperationDelete)
// waits for the owner backfill, or a restart finds hydrant with work leftfunc (w *world) seed(n *node, collaborators ...syntax.DID) { t := w.t t.Helper() w.serve(testRepoDid, "repo", collaborators...) w.emitDeclaration("repo", testRepoDid) last := w.emitManifest(testRepoDid) eventually(t, 15*time.Second, "the repo to be served", func() bool { return len(repoRows(t, n.s, testRepoDid)) == 1 }) for _, c := range collaborators { eventually(t, 15*time.Second, "the collaborators to be synced", func() bool { return isCollaborator(t, n.s, testRepoDid, c) }) } eventually(t, 15*time.Second, "hydrant to backfill the owner", func() bool { return w.hyd.repoStatus(testOwner) == synced(testOwner).Status }) n.consumed(last)}
// a restart rewinds the cursor a second, so push it past everything firstfunc (w *world) pushCursorPast(n *node) { t := w.t t.Helper() at := w.js.filler() n.consumed(at) n.stop() if n.cursor() <= at { t.Fatalf("setup: the saved cursor %d never reached the filler event at %d", n.cursor(), at) }}
func TestShortOutageReplaysFromTheSavedCursor(t *testing.T) { w := newWorld(t) n := w.boot() w.seed(n, testBob, testCarol) n.stop() erased := len(w.hyd.erasedDids())
w.addCollaborator(testRepoDid, testDave) w.emitCollaborator(testRepoDid, testDave, opCreate) w.removeCollaborator(testRepoDid, testCarol) w.emitCollaborator(testRepoDid, testCarol, opDelete) w.setManifest(testRepoDid, "Repo", w.spindleDid()) w.emitManifest(testRepoDid)
n2 := w.boot() eventually(t, 15*time.Second, "the replay to bring back the collaborators and the manifest", func() bool { rows := repoRows(t, n2.s, testRepoDid) return isCollaborator(t, n2.s, testRepoDid, testDave) && !isCollaborator(t, n2.s, testRepoDid, testCarol) && len(rows) == 1 && rows[0].Name == "Repo" }) if !isCollaborator(t, n2.s, testRepoDid, testBob) { t.Error("bob lost access during the replay") } if got := w.hyd.erasedDids(); len(got) != erased { t.Errorf("a catch-up ran for a short outage, hydrant erased %v", got[erased:]) }}
func TestLongOutageCatchUpBringsBackWhatChangedWhileDown(t *testing.T) { scenarios := []struct { name string change func(w *world) check func(t *testing.T, n *node) bool }{ { name: "collaborator added", change: func(w *world) { w.addCollaborator(testRepoDid, testDave) }, check: func(t *testing.T, n *node) bool { return isCollaborator(t, n.s, testRepoDid, testDave) && isCollaborator(t, n.s, testRepoDid, testBob) }, }, { name: "collaborator removed", change: func(w *world) { w.removeCollaborator(testRepoDid, testBob) }, check: func(t *testing.T, n *node) bool { if isCollaborator(t, n.s, testRepoDid, testBob) { return false } if ok, _ := n.s.e.IsSettingsAllowed(testBob.String(), rbac.ThisServer, testRepoDid.String()); ok { t.Error("a removed collaborator can still reach the repo's secrets") } if secretCount(t, n.s, testRepoDid) != 1 { t.Error("removing a collaborator dropped the secrets") } return isCollaborator(t, n.s, testRepoDid, testCarol) }, }, { name: "manifest switched to another spindle", change: func(w *world) { w.setManifest(testRepoDid, "repo", "did:web:other.example.com") }, check: func(t *testing.T, n *node) bool { if len(repoRows(t, n.s, testRepoDid)) != 0 { return false } if ok, _ := n.s.e.IsRepoOwner(testOwner.String(), rbac.ThisServer, testRepoDid.String()); ok { t.Error("the owner kept access to a repo this spindle forgot") } if isCollaborator(t, n.s, testRepoDid, testBob) { t.Error("a collaborator kept access to a repo this spindle forgot") } if secretCount(t, n.s, testRepoDid) != 1 { t.Error("forgetting a repo dropped its secrets, they have to stay for a move back") } return true }, }, { name: "manifest switched to this spindle", change: func(w *world) { w.setManifest(faultOtherRepoDid, "other", w.spindleDid()) }, check: func(t *testing.T, n *node) bool { return len(repoRows(t, n.s, faultOtherRepoDid)) == 1 && isCollaborator(t, n.s, faultOtherRepoDid, testBob) }, }, { name: "repo deleted", change: func(w *world) { w.knot.erase(testRepoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey) w.removeCollaborator(testRepoDid, testBob) w.removeCollaborator(testRepoDid, testCarol) w.undeclare("repo") }, check: func(t *testing.T, n *node) bool { if len(repoRows(t, n.s, testRepoDid)) != 0 { return false } if secretCount(t, n.s, testRepoDid) != 0 { t.Error("a deleted repo kept its secrets") } return true }, }, { name: "repo renamed", change: func(w *world) { w.setManifest(testRepoDid, "New-Name", w.spindleDid()) w.undeclare("repo") w.declare("New-Name", testRepoDid) }, check: func(t *testing.T, n *node) bool { rows := repoRows(t, n.s, testRepoDid) if len(rows) != 1 || rows[0].Rkey != "new-name" { return false } if rows[0].Name != "New-Name" || !isCollaborator(t, n.s, testRepoDid, testBob) || secretCount(t, n.s, testRepoDid) != 1 { t.Errorf("rename lost something, rows = %s, bob = %v, secrets = %d", describeRows(rows), isCollaborator(t, n.s, testRepoDid, testBob), secretCount(t, n.s, testRepoDid)) } return true }, }, }
for _, sc := range scenarios { t.Run(sc.name, func(t *testing.T) { w := newWorld(t) w.setManifest(faultOtherRepoDid, "other", "did:web:other.example.com") w.declare("other", faultOtherRepoDid) w.addCollaborator(faultOtherRepoDid, testBob)
n := w.boot() w.seed(n, testBob, testCarol) addTestSecret(t, n.s, testRepoDid) if len(repoRows(t, n.s, faultOtherRepoDid)) != 0 { t.Fatal("setup: the other repo is served already") } n.stop()
// jetstream forgets, so only hydrant can bring these changes back sc.change(w) w.js.forget() w.js.filler()
n2 := w.boot() eventually(t, 20*time.Second, "the catch-up to reach hydrant", func() bool { return slices.Contains(w.hyd.erasedDids(), testRepoDid) }) eventually(t, 20*time.Second, "the catch-up to bring back what changed", func() bool { return sc.check(t, n2) }) }) }}
func TestAnEventCutShortByShutdownIsNotLost(t *testing.T) { w := newWorld(t) n := w.boot() // a first boot with no saved cursor connects live, so save one first n.consumed(w.js.filler())
w.serve(testRepoDid, "repo", testBob) release := w.holdManifest() defer release(false) w.emitManifest(testRepoDid) eventually(t, 10*time.Second, "the event to start its refresh", func() bool { return w.heldReads.Load() > 0 })
n.stop() release(false) if len(repoRows(t, n.s, testRepoDid)) != 0 { t.Fatal("setup: the cut event was handled anyway") }
n2 := w.boot() eventually(t, 15*time.Second, "the cut event to be replayed", func() bool { return len(repoRows(t, n2.s, testRepoDid)) == 1 && isCollaborator(t, n2.s, testRepoDid, testBob) })}
func TestSecretsSurviveEveryWayTheWorldCanLookBroken(t *testing.T) { w := newWorld(t) n := w.boot() w.seed(n, testBob) addTestSecret(t, n.s, testRepoDid)
faults := []struct { name string set func(on bool) }{ {"bare 404 from a proxy", w.setBare404}, {"did document names another knot", func(on bool) { knot := w.knot if on { knot = w.other } w.plc.setKnot(testRepoDid, knot.srv.URL) n.s.res.InvalidateIdent(context.Background(), testRepoDid.String()) }}, {"declaration missing", func(on bool) { if on { w.undeclare("repo") } else { w.declare("repo", testRepoDid) } }}, {"pds error", w.setPDSFailing}, {"knot outage", w.setKnotDown}, } for i, f := range faults { f.set(true) before := w.traffic() w.emitManifest(testRepoDid) eventually(t, 10*time.Second, "spindle to look at the repo", func() bool { return w.traffic() > before }) w.quiet() f.set(false)
// events run in order, so the marker landing means the broken one is done marker := syntax.DID("did:plc:marker" + strings.Repeat("x", i+1)) w.addCollaborator(testRepoDid, marker) w.emitCollaborator(testRepoDid, marker, opCreate) eventually(t, 20*time.Second, "the next change to be applied", func() bool { return isCollaborator(t, n.s, testRepoDid, marker) })
if rows := repoRows(t, n.s, testRepoDid); len(rows) != 1 { t.Errorf("%s: rows = %s", f.name, describeRows(rows)) } if got := secretCount(t, n.s, testRepoDid); got != 1 { t.Errorf("%s: %d secrets left, want 1", f.name, got) } if !isCollaborator(t, n.s, testRepoDid, testBob) { t.Errorf("%s: bob lost access", f.name) } }}
func TestRefreshThatFailedAfterALiveEventIsRetriedAfterARestart(t *testing.T) { w := newWorld(t) n := w.boot() w.seed(n, testBob)
w.setKnotDown(true) before := w.traffic() w.addCollaborator(testRepoDid, testCarol) w.emitCollaborator(testRepoDid, testCarol, opCreate) eventually(t, 10*time.Second, "the refresh to be attempted", func() bool { return w.traffic() > before }) w.quiet() w.pushCursorPast(n)
// retry backoff is too long to wait out, so skew the clock forward w.setKnotDown(false) w.skew.Store(int64(time.Hour)) n2 := w.boot() eventually(t, 40*time.Second, "carol to get access after the restart", func() bool { return isCollaborator(t, n2.s, testRepoDid, testCarol) })}
func TestCatchUpInterruptedByARestartIsNotForgotten(t *testing.T) { w := newWorld(t) n := w.boot() w.seed(n, testBob) n.stop()
w.addCollaborator(testRepoDid, testCarol) w.emitCollaborator(testRepoDid, testCarol, opCreate) w.js.forget() w.js.filler() w.hyd.failStartsWith(errors.New("hydrant cannot start"))
n2 := w.boot() w.pushCursorPast(n2)
w.hyd.failStartsWith(nil) n3 := w.boot() eventually(t, 40*time.Second, "the catch-up to run on the next boot", func() bool { return isCollaborator(t, n3.s, testRepoDid, testCarol) })}
func TestPartlyFailedCatchUpIsRetriedOnTheNextBoot(t *testing.T) { w := newWorld(t) n := w.boot() w.seed(n, testBob) n.stop()
// only the repo backfill can see this, the old owner's declaration is gone w.knot.put(testRepoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(testNewOwner, "repo", w.spindleDid())) w.undeclare("repo") w.knot.declare(testNewOwner, "repo", testRepoDid) w.plc.setPDS(testNewOwner, w.knot.srv.URL) w.js.forget() w.js.filler() w.setHydrantFailing(testRepoDid, true)
n2 := w.boot() eventually(t, 20*time.Second, "hydrant to give up on the repo", func() bool { return w.hyd.repoStatus(testRepoDid) == failedWith(testRepoDid, "transport", 0).Status }) if rows := repoRows(t, n2.s, testRepoDid); len(rows) != 1 || rows[0].Owner != testOwner { t.Fatalf("setup: the move was noticed without the repo's backfill, rows = %s", describeRows(rows)) } w.pushCursorPast(n2)
w.setHydrantFailing(testRepoDid, false) erased := len(w.hyd.erasedDids()) n3 := w.boot() eventually(t, 40*time.Second, "the repo to be backfilled on the next boot", func() bool { rows := repoRows(t, n3.s, testRepoDid) return len(rows) == 1 && rows[0].Owner == testNewOwner }) if got := w.hyd.erasedDids()[erased:]; slices.Contains(got, testOwner) { t.Errorf("the owner backfilled in the first catch-up but was redone, erased %v", got) }}
type pushEnv struct { w *world n *node repo *fakeRepo}
func newPushEnv(t *testing.T) *pushEnv { t.Helper() w := newWorld(t) repo := w.serveGitRepo(testRepoDid) n := w.boot() w.seed(n, testBob) return &pushEnv{w: w, n: n, repo: repo}}
func faultPush(ref, oldSha, newSha string) *org_tangled.EventNotifyEvent_Input_Event { actor := testBob.String() return &org_tangled.EventNotifyEvent_Input_Event{EventPush: &org_tangled.EventPush{ Repo: testRepoDid.String(), Ref: ref, OldSha: oldSha, NewSha: newSha, Actor: &actor, }}}
var zeroSha = strings.Repeat("0", 40)
func (e *pushEnv) notify(s *Spindle, event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) { return s.NotifyEvent(context.Background(), e.w.knotRef(), testRepoDid, event)}
func (e *pushEnv) firstPush(t *testing.T) *org_tangled.EventNotifyEvent_Input_Event { t.Helper() return faultPush("refs/heads/main", zeroSha, e.repo.commit(t))}
func (e *pushEnv) waitClone(t *testing.T) { t.Helper() eventually(t, 10*time.Second, "the clone to start", func() bool { return e.w.gitFetches.Load() > 0 })}
func (e *pushEnv) wantPipelines(t *testing.T, want int, when string) { t.Helper() if got := e.n.pipelines(testRepoDid); got != want { t.Fatalf("%d pipelines %s, want %d", got, when, want) }}
func (e *pushEnv) mustNotify(t *testing.T, event *org_tangled.EventNotifyEvent_Input_Event) models.PipelineId { t.Helper() uri, err := e.notify(e.n.s, event) if err != nil { t.Fatalf("NotifyEvent: %v", err) } return uri}
type notifyAnswer struct { uri models.PipelineId err error}
func (e *pushEnv) notifyAsync(ctx context.Context, s *Spindle, event *org_tangled.EventNotifyEvent_Input_Event) <-chan notifyAnswer { answers := make(chan notifyAnswer, 1) go func() { uri, err := s.NotifyEvent(ctx, e.w.knotRef(), testRepoDid, event) answers <- notifyAnswer{uri, err} }() return answers}
func awaitAnswer(t *testing.T, answers <-chan notifyAnswer) notifyAnswer { t.Helper() select { case a := <-answers: return a case <-time.After(30 * time.Second): t.Fatal("NotifyEvent never answered") return notifyAnswer{} }}
func TestNotifyDoesNotAnswerBeforeThePushIsHandled(t *testing.T) { e := newPushEnv(t) push := e.firstPush(t)
release := e.w.holdGit() defer release(false) answers := e.notifyAsync(context.Background(), e.n.s, push) e.waitClone(t)
// the knot waits on this call, an early answer drops a push that then fails select { case a := <-answers: t.Fatalf("answered %q, %v while the push was still being cloned", a.uri, a.err) case <-time.After(3 * time.Second): }
release(true) if a := awaitAnswer(t, answers); a.err == nil { t.Fatalf("answered %q without an error for a push that failed", a.uri) } e.wantPipelines(t, 0, "after a failed push")
if uri := e.mustNotify(t, push); uri == "" { t.Fatal("the retry made no pipeline") } e.wantPipelines(t, 1, "after the retry")}
func TestNotifyRetryAfterACrashBeforeThePipelineRowRunsThePush(t *testing.T) { e := newPushEnv(t) push := e.firstPush(t)
release := e.w.holdGit() defer release(true) ctx, cancel := context.WithCancel(context.Background()) answers := e.notifyAsync(ctx, e.n.s, push) e.waitClone(t)
e.n.stop() cancel() if a := awaitAnswer(t, answers); a.err == nil { t.Fatalf("answered %q without an error although spindle went down first", a.uri) } release(true)
n2 := e.w.boot() if got := n2.pipelines(testRepoDid); got != 0 { t.Fatalf("%d pipelines before the knot's retry", got) } uri, err := e.notify(n2.s, push) if err != nil || uri == "" { t.Fatalf("the knot's retry got %q, %v", uri, err) } if got := n2.pipelines(testRepoDid); got != 1 { t.Fatalf("%d pipelines after the retry, want 1", got) }}
func TestNotifyReplayWhileTheFirstRunIsInFlightJoinsIt(t *testing.T) { e := newPushEnv(t) push := e.firstPush(t)
release := e.w.holdGit() defer release(false) first := e.notifyAsync(context.Background(), e.n.s, push) e.waitClone(t) second := e.notifyAsync(context.Background(), e.n.s, push) time.Sleep(300 * time.Millisecond) release(false)
a, b := awaitAnswer(t, first), awaitAnswer(t, second) if a.err != nil || b.err != nil || a.uri == "" || a.uri != b.uri { t.Fatalf("answers %+v and %+v, want one pipeline for both", a, b) } e.wantPipelines(t, 1, "after a replay joined the run in flight")}
func TestNotifyReplayAfterItFinishedIsSkipped(t *testing.T) { // a restart forgets everything in memory, so that answer has to come from the db for _, restart := range []bool{false, true} { t.Run(fmt.Sprintf("restart=%v", restart), func(t *testing.T) { e := newPushEnv(t) push := e.firstPush(t) first := e.mustNotify(t, push) if first == "" { t.Fatal("the push made no pipeline") } fetches := e.w.gitFetches.Load() if restart { e.n.stop() e.n = e.w.boot() }
if again := e.mustNotify(t, push); again != first { t.Fatalf("the replay answered %q, want %q", again, first) } e.wantPipelines(t, 1, "after a replay") if got := e.w.gitFetches.Load(); got != fetches { t.Fatalf("the replay cloned again, %d fetches after %d", got, fetches) } }) }}
func TestNotifyDedupeIsPerRepoRefOldAndNew(t *testing.T) { e := newPushEnv(t) c1, c2 := e.repo.commit(t), e.repo.commit(t) main := "refs/heads/main"
created := e.mustNotify(t, faultPush(main, zeroSha, c1)) moved := e.mustNotify(t, faultPush(main, c1, c2)) // a revert to an old commit is a new push and still runs back := e.mustNotify(t, faultPush(main, c2, c1)) if created == "" || moved == "" || back == "" || created == moved || moved == back || created == back { t.Fatalf("pipelines %q, %q, %q, want three different ones", created, moved, back) } e.wantPipelines(t, 3, "after three pushes")
// the same old and new twice is skipped on purpose if again := e.mustNotify(t, faultPush(main, c1, c2)); again != moved { t.Fatalf("the repeated push answered %q, want %q", again, moved) } e.wantPipelines(t, 3, "after the repeat")}
func TestNotifyDeleteAndRecreateRunsAgain(t *testing.T) { e := newPushEnv(t) c0, c1 := e.repo.commit(t), e.repo.commit(t) main := "refs/heads/main"
e.mustNotify(t, faultPush(main, c0, c1)) if gone := e.mustNotify(t, faultPush(main, c1, zeroSha)); gone != "" { t.Fatalf("a deleted branch answered %q", gone) } if again := e.mustNotify(t, faultPush(main, zeroSha, c1)); again == "" { t.Fatal("the branch came back at a commit that already ran and nothing ran") } e.wantPipelines(t, 2, "after the recreate")}