package spindle
import (
"context"
"fmt"
"net/http"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/knotclient"
"tangled.org/core/tapc"
)
func didSharingStripe(with syntax.DID) syntax.DID {
for i := 0; ; i++ {
did := syntax.DID(fmt.Sprintf("did:plc:share%d", i))
if did != with && stripeOf(did) == stripeOf(with) {
return did
}
}
}
func goErr(fn func() error) <-chan error {
done := make(chan error, 1)
go func() { done <- fn() }()
return done
}
func waitErr(t *testing.T, done <-chan error, within time.Duration, timedOut string) error {
t.Helper()
select {
case err := <-done:
return err
case <-time.After(within):
t.Fatal(timedOut)
return nil
}
}
func ensureAsync(env *ensureEnv) <-chan error {
return goErr(func() error {
_, err := env.s.ensureRepo(context.Background(), env.k, testRepoDid)
return err
})
}
func pullEventRows(t *testing.T, s *Spindle, repo syntax.DID) int {
t.Helper()
var n int
if err := s.db.QueryRow("select count(*) from pull_events where repo_did = ?", repo.String()).Scan(&n); err != nil {
t.Fatal(err)
}
return n
}
func hungPatchHandler(t *testing.T) (e *pullEnv, done <-chan error, release func()) {
t.Helper()
e = newPullEnv(t)
e.putTicket(testRepoDid, pullRkey, "open", testBob, true)
e.putPatch(testRepoDid, patchOne, pullRkey, patchSha1, testBob)
if repo, err := e.n.s.servedRepo(testRepoDid); err != nil || repo == nil {
t.Fatalf("setup: the repo isn't served: %v", err)
}
rec := &tapc.RecordEventData{
Did: testRepoDid,
Collection: syntax.NSID(org_tangled.ReviewPatchNSID),
Rkey: syntax.RecordKey(patchOne),
Action: tapc.RecordCreateAction,
Live: true,
}
retry := db.RecordRetry{RepoDid: testRepoDid, Collection: org_tangled.ReviewPatchNSID, Rkey: patchOne}
entered, gate := make(chan struct{}), make(chan struct{})
done = goErr(func() error {
return e.n.s.applyPullRecord(context.Background(), rec, retry, func(ctx context.Context, repo *db.Repo, rec *tapc.RecordEventData) error {
close(entered)
<-gate
return e.n.s.applyPatch(ctx, repo, rec)
})
})
var once sync.Once
release = func() { once.Do(func() { close(gate) }) }
t.Cleanup(release)
select {
case <-entered:
case <-time.After(5 * time.Second):
t.Fatal("the handler never started")
}
return e, done, release
}
func TestRepoLockExcludesTheSameDid(t *testing.T) {
s := &Spindle{}
a := syntax.DID("did:plc:locka")
unlock := s.lockRepo(a).Unlock
got := make(chan struct{})
go func() {
s.lockRepo(a).Unlock()
close(got)
}()
select {
case <-got:
t.Fatal("a second lock on the same repo was granted while the first was held")
case <-time.After(200 * time.Millisecond):
}
unlock()
select {
case <-got:
case <-time.After(time.Second):
t.Fatal("the second lock was never granted after the first was released")
}
}
func hungPush(t *testing.T) (e *pushEnv, release func(fail bool), answers <-chan notifyAnswer) {
t.Helper()
e = newPushEnv(t)
release = e.w.holdGit()
answers = e.notifyAsync(context.Background(), e.n.s, e.firstPush(t))
e.waitClone(t)
return e, release, answers
}
func TestHungFetchDoesNotDelayAnUnrelatedRepo(t *testing.T) {
e, release, answers := hungPush(t)
defer release(false)
other := didSharingStripe(testRepoDid)
row := db.Repo{Knot: e.w.knotRef().Host, Owner: testOwner, Rkey: "other", RepoDid: other, Name: "other"}
if err := e.n.s.db.AddRepo(row); err != nil {
t.Fatal(err)
}
push := faultPush("refs/heads/main", zeroSha, strings.Repeat("b", 40))
push.EventPush.PushOptions = []string{"skip-ci"}
done := goErr(func() error {
_, err := e.n.s.NotifyEvent(context.Background(), e.w.knotRef(), other, push)
return err
})
if err := waitErr(t, done, 3*time.Second, "a notify for an unrelated repo took more than 3s while another repo's fetch hung"); err != nil {
t.Fatalf("notify for the unrelated repo: %v", err)
}
release(false)
awaitAnswer(t, answers)
}
func TestHungFetchDoesNotDelayARefreshOfTheSameRepo(t *testing.T) {
e, release, answers := hungPush(t)
defer release(false)
done := goErr(func() error {
_, err := e.n.s.ensureRepo(context.Background(), e.w.knotRef(), testRepoDid)
return err
})
if err := waitErr(t, done, 3*time.Second, "a refresh of a repo took more than 3s while its own push's fetch hung"); err != nil {
t.Fatalf("refresh of the repo: %v", err)
}
release(false)
awaitAnswer(t, answers)
}
func TestPipelineInsertWaitsForTheRepoLock(t *testing.T) {
e, release, answers := hungPush(t)
defer release(false)
locked := make(chan func(), 1)
go func() { locked <- e.n.s.lockRepo(testRepoDid).Unlock }()
var unlock func()
select {
case unlock = <-locked:
case <-time.After(3 * time.Second):
t.Fatal("the repo lock couldn't be taken while a fetch for the repo hung")
}
release(false)
select {
case a := <-answers:
unlock()
t.Fatalf("the push answered %q, %v while the repo lock was held", a.uri, a.err)
case <-time.After(1500 * time.Millisecond):
}
e.wantPipelines(t, 0, "while the repo lock was held")
unlock()
if a := awaitAnswer(t, answers); a.err != nil || a.uri == "" {
t.Fatalf("the push answered %q, %v after the lock was released, want a pipeline", a.uri, a.err)
}
e.wantPipelines(t, 1, "after the repo lock was released")
}
func TestConcurrentEnrolmentMakesOneRow(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo", testBob)
var wg sync.WaitGroup
errs := make(chan error, 8)
for range 8 {
wg.Go(func() {
_, err := env.s.ensureRepo(context.Background(), env.k, testRepoDid)
errs <- err
})
}
wg.Wait()
close(errs)
for err := range errs {
if err != nil {
t.Fatalf("ensureRepo: %v", err)
}
}
if rows := env.rows(t); len(rows) != 1 {
t.Fatalf("%d repo rows after 8 concurrent enrolments, want 1: %+v", len(rows), rows)
}
collaborators, err := env.s.db.ListCollaboratorsByRepoDid(testRepoDid)
if err != nil || len(collaborators) != 1 {
t.Fatalf("collaborators = %+v, %v, want exactly one", collaborators, err)
}
}
func TestPullHandlerHeldMidIoLeavesNoRowsAfterAWipe(t *testing.T) {
e, done, release := hungPatchHandler(t)
if err := e.n.s.WipeRepo(context.Background(), testRepoDid, "test"); err != nil {
t.Fatalf("wipe: %v", err)
}
release()
waitErr(t, done, 10*time.Second, "the handler never finished")
for _, table := range []string{"pull_tickets", "pull_patches", "pull_events"} {
var n int
if err := e.n.s.db.QueryRow("select count(*) from "+table+" where repo_did = ?", testRepoDid.String()).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Errorf("%d rows in %s for a wiped repo, want 0", n, table)
}
}
}
type gatedTransport struct {
match func(*http.Request) bool
// answers before the gate, and the caller gets the old reply
afterwards bool
first atomic.Bool
entered chan struct{}
gate chan struct{}
}
func (g *gatedTransport) RoundTrip(r *http.Request) (*http.Response, error) {
if !g.match(r) || !g.first.CompareAndSwap(false, true) {
return http.DefaultTransport.RoundTrip(r)
}
var resp *http.Response
if g.afterwards {
var err error
if resp, err = http.DefaultTransport.RoundTrip(r); err != nil {
return nil, err
}
}
close(g.entered)
select {
case <-g.gate:
case <-r.Context().Done():
return nil, r.Context().Err()
}
if resp != nil {
return resp, nil
}
return http.DefaultTransport.RoundTrip(r)
}
func TestStaleRefreshDoesNotOverwriteANewerOne(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
if env.ensure(t) == nil {
t.Fatal("setup: the repo isn't served")
}
gate := &gatedTransport{
match: func(r *http.Request) bool {
return strings.Contains(r.URL.RawQuery, "collection="+org_tangled.RepoDeclarationNSID)
},
entered: make(chan struct{}),
gate: make(chan struct{}),
}
env.s.knots = knotclient.New(env.s.res.Directory(), &http.Client{Transport: gate}, true)
first := ensureAsync(env)
select {
case <-gate.entered:
case <-time.After(5 * time.Second):
t.Fatal("the first refresh never reached its declaration read")
}
env.manifest("repo", "did:web:other.example.com")
second := ensureAsync(env)
// the second can finish alone or wait on the first, either is fine
time.Sleep(2 * time.Second)
close(gate.gate)
if err := waitErr(t, first, 5*time.Second, "the first refresh never finished"); err != nil {
t.Fatalf("first refresh: %v", err)
}
if err := waitErr(t, second, 5*time.Second, "the second refresh never finished"); err != nil {
t.Fatalf("second refresh: %v", err)
}
if rows := env.rows(t); len(rows) != 0 {
t.Fatalf("repo rows = %+v, want none: the refresh that read the old manifest was applied after the newer one", rows)
}
}
func TestStaleForgetDoesNotUndoANewerRegister(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
if env.ensure(t) == nil {
t.Fatal("setup: the repo isn't served")
}
gate := &gatedTransport{
match: func(r *http.Request) bool {
return strings.Contains(r.URL.RawQuery, "collection="+org_tangled.RepoManifestNSID)
},
afterwards: true,
entered: make(chan struct{}),
gate: make(chan struct{}),
}
env.manifest("repo", "did:web:other.example.com")
env.s.knots = knotclient.New(env.s.res.Directory(), &http.Client{Transport: gate}, true)
first := ensureAsync(env)
select {
case <-gate.entered:
case <-time.After(5 * time.Second):
t.Fatal("the first refresh never read the manifest")
}
env.manifest("repo", env.spindleDid())
if _, err := env.s.ensureRepo(context.Background(), env.k, testRepoDid); err != nil {
t.Fatalf("second refresh: %v", err)
}
close(gate.gate)
if err := waitErr(t, first, 5*time.Second, "the first refresh never finished"); err != nil {
t.Fatalf("first refresh: %v", err)
}
if rows := env.rows(t); len(rows) != 1 {
t.Fatalf("%d repo rows, want 1: the refresh that read the old manifest forgot a repo a newer one had kept", len(rows))
}
}
// turns are ordered per did, sharing a stripe only shares the mutex
func TestTurnsOfDidsSharingAStripeBothApply(t *testing.T) {
env := newEnsureEnv(t)
a := syntax.DID("did:plc:stripea")
b := didSharingStripe(a)
older, newer := env.s.beginTurn(a), env.s.beginTurn(b)
for name, turn := range map[string]*repoTurn{"newer": newer, "older": older} {
applied, err := turn.apply(func() error { return nil })
if err != nil || !applied {
t.Errorf("the %s turn applied=%v err=%v, want it applied", name, applied, err)
}
}
}
func TestOlderTurnOfTheSameDidIsSkipped(t *testing.T) {
env := newEnsureEnv(t)
a := syntax.DID("did:plc:turns")
older, newer := env.s.beginTurn(a), env.s.beginTurn(a)
if applied, err := newer.apply(func() error { return nil }); err != nil || !applied {
t.Fatalf("the newer turn applied=%v err=%v", applied, err)
}
if applied, err := newer.apply(func() error { return nil }); err != nil || !applied {
t.Fatalf("a second write of the same turn applied=%v err=%v, want it applied", applied, err)
}
ran := false
if applied, err := older.apply(func() error { ran = true; return nil }); err != nil || applied || ran {
t.Fatalf("the older turn applied=%v ran=%v err=%v, want it skipped", applied, ran, err)
}
}
func TestHungPullHandlerDoesNotDelayAPushOrAPullEventOfAStripeNeighbour(t *testing.T) {
e, _, _ := hungPatchHandler(t)
neighbour := didSharingStripe(testRepoDid)
row := db.Repo{Knot: e.knotRef().Host, Owner: testOwner, Rkey: "neighbour", RepoDid: neighbour, Name: "neighbour"}
if err := e.n.s.db.AddRepo(row); err != nil {
t.Fatal(err)
}
push := faultPush("refs/heads/main", zeroSha, strings.Repeat("b", 40))
push.EventPush.PushOptions = []string{"skip-ci"}
pushed := goErr(func() error {
_, err := e.n.s.NotifyEvent(context.Background(), e.knotRef(), neighbour, push)
return err
})
if err := waitErr(t, pushed, 3*time.Second, "a push for a repo on the same stripe took more than 3s while a pull handler hung on the knot"); err != nil {
t.Fatalf("push for the neighbour: %v", err)
}
patch := db.PullPatch{Rkey: "p1", Ticket: "t1", Head: patchSha1, CreatedAt: "2026-02-01T00:00:00Z"}
ev := db.PullEvent{RepoDid: neighbour, Ticket: "t1", Author: testBob, Sender: testBob, Pull: []byte("{}"), Pipeline: "p1"}
if err := e.n.s.db.AddPullPatch(neighbour, patch, ev); err != nil {
t.Fatal(err)
}
e.n.s.retries.jobs.kick(neighbour)
eventually(t, 5*time.Second, "the neighbour's pull event to run while a pull handler hung", func() bool {
return pullEventRows(t, e.n.s, neighbour) == 0
})
}
func TestKicksDuringAPullEventDoNotRunItTwice(t *testing.T) {
e := newPushEnv(t)
sink := newSink(t)
e.w.subscribe(e.n, sink, pullEvents...)
sha := e.repo.commit(t)
e.w.putTicket(testRepoDid, pullRkey, "open", testBob, true)
e.w.putPatch(testRepoDid, patchOne, pullRkey, sha, testBob)
release := e.w.holdGit()
defer release(false)
e.w.emitTicket(testRepoDid, pullRkey, opCreate)
e.w.emitPatch(testRepoDid, patchOne, opCreate)
e.waitClone(t)
e.n.s.retries.jobs.kick(testRepoDid)
e.n.s.retries.jobs.kick(testRepoDid)
e.n.s.kickDueRepos()
time.Sleep(500 * time.Millisecond)
release(false)
eventually(t, 10*time.Second, "the pull event to finish", func() bool { return pullEventRows(t, e.n.s, testRepoDid) == 0 })
sink.expect(t, 1)
}
// the stored turn sits an hour ahead, like after a clock stepped back
func TestATurnAfterARestartIsNotBehindTheStoredOnes(t *testing.T) {
s := newWipeTestSpindle(t)
a := syntax.DID("did:plc:turns")
ahead := time.Now().Add(time.Hour).UnixNano()
if _, err := s.db.AdvanceRepoTurn(a, ahead, time.Now()); err != nil {
t.Fatal(err)
}
applied, err := s.beginTurn(a).apply(func() error { return nil })
if err != nil || !applied {
t.Fatalf("the first turn after a restart applied=%v err=%v, want it applied over a stored one from the future", applied, err)
}
}