package 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 until spindle has finished the setup work, so a restart doesn't find some of it left over
func (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.applied(testRepoDid, last)
}
// a restart rewinds the cursor a second, so push it past everything first
func (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.deleteRepo(testRepoDid)
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)
// skip the backoff the failed refresh left behind
w.skew.Add(int64(2 * time.Hour))
// 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 })
// wait until the catch-up is done with everything but the repo, which is left for the next boot
eventually(t, 20*time.Second, "the catch-up to keep only the repo pending", func() bool {
pending, err := n2.s.db.PendingCatchUp()
return err == nil && slices.Equal(pending, []syntax.DID{testRepoDid})
})
w.pushCursorPast(n2)
w.setHydrantFailing(testRepoDid, false)
erased := len(w.hyd.erasedDids())
n3 := w.boot()
eventually(t, 40*time.Second, "the repo's backfill to be retried on the next boot", func() bool { return w.hyd.repoStatus(testRepoDid) == synced(testRepoDid).Status })
eventually(t, 40*time.Second, "the repo to have moved", 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")
}