package spindle
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/jetstream"
)
type ticketLog struct {
mu sync.Mutex
done []string
dropped []string
}
func (l *ticketLog) snapshot() (done, dropped []string) {
l.mu.Lock()
defer l.mu.Unlock()
return slices.Clone(l.done), slices.Clone(l.dropped)
}
type namedTicket struct {
name string
log *ticketLog
}
func (t namedTicket) Done() {
t.log.mu.Lock()
defer t.log.mu.Unlock()
t.log.done = append(t.log.done, t.name)
}
func (t namedTicket) Drop() {
t.log.mu.Lock()
defer t.log.mu.Unlock()
t.log.dropped = append(t.log.dropped, t.name)
}
func (sh *ingestShards) wait() { sh.wg.Wait() }
func dispatchAll(t *testing.T, ctx context.Context, sh *ingestShards, log *ticketLog, did string, rkeys ...string) {
t.Helper()
for _, rkey := range rkeys {
if err := sh.dispatch(ctx, shardEvent(did, rkey), namedTicket{rkey, log}); err != nil {
t.Fatal(err)
}
}
}
func ticketDone(log *ticketLog, name string) bool {
done, _ := log.snapshot()
return slices.Contains(done, name)
}
func shardEvent(did, rkey string) *jetstream.Event {
return &jetstream.Event{Did: did, Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
Operation: jetstream.CommitOperationCreate, Collection: org_tangled.GraphCollaboratorNSID, RKey: rkey,
}}
}
func otherShardKey(key string) string {
for i := 0; ; i++ {
if other := fmt.Sprintf("did:plc:other%d", i); ingestShardOf(other) != ingestShardOf(key) {
return other
}
}
}
type appliedLog struct {
mu sync.Mutex
seen []string
}
func (a *appliedLog) add(s string) {
a.mu.Lock()
defer a.mu.Unlock()
a.seen = append(a.seen, s)
}
func (a *appliedLog) has(s string) bool {
a.mu.Lock()
defer a.mu.Unlock()
return slices.Contains(a.seen, s)
}
func (a *appliedLog) list() []string {
a.mu.Lock()
defer a.mu.Unlock()
return slices.Clone(a.seen)
}
func TestShardsKeepARepoInOrderWhileOtherReposGoOn(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
var applied appliedLog
gate := make(chan struct{})
sh := newIngestShards(ctx, func(_ context.Context, e *jetstream.Event) error {
if e.Commit.RKey == "x1" {
<-gate
}
applied.add(e.Commit.RKey)
return nil
})
t.Cleanup(func() { cancel(); sh.wait() })
x := "did:plc:x"
y := otherShardKey(x)
var log ticketLog
dispatchAll(t, ctx, sh, &log, x, "x1", "x2")
dispatchAll(t, ctx, sh, &log, y, "y1", "y2")
eventually(t, 2*time.Second, "the other repo to apply while x1 is stuck", func() bool { return applied.has("y2") })
time.Sleep(50 * time.Millisecond)
if applied.has("x1") || applied.has("x2") {
t.Fatalf("x ran ahead of its stuck first event, applied = %v", applied.list())
}
if ticketDone(&log, "x1") || ticketDone(&log, "x2") {
t.Fatal("a ticket finished before its event applied")
}
close(gate)
eventually(t, 2*time.Second, "x to finish", func() bool { return applied.has("x2") })
var xs []string
for _, rkey := range applied.list() {
if rkey[0] == 'x' {
xs = append(xs, rkey)
}
}
if !slices.Equal(xs, []string{"x1", "x2"}) {
t.Fatalf("x applied in the order %v, want x1 then x2", xs)
}
eventually(t, 2*time.Second, "every ticket to finish", func() bool {
done, _ := log.snapshot()
return len(done) == 4
})
}
func TestShardsDropEveryTicketTheyDidNotFinishAtShutdown(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
started := make(chan struct{})
sh := newIngestShards(ctx, func(ctx context.Context, e *jetstream.Event) error {
if e.Commit.RKey == "x1" {
close(started)
<-ctx.Done()
return ctx.Err()
}
return nil
})
x := "did:plc:x"
var log ticketLog
dispatchAll(t, ctx, sh, &log, x, "x1", "x2", "x3")
<-started
cancel()
sh.wait()
done, dropped := log.snapshot()
slices.Sort(dropped)
if len(done) != 0 || !slices.Equal(dropped, []string{"x1", "x2", "x3"}) {
t.Fatalf("done = %v, dropped = %v, want all three dropped and none finished", done, dropped)
}
err := sh.dispatch(context.Background(), shardEvent(x, "x4"), namedTicket{"x4", &log})
if err == nil {
t.Fatal("a stopped shard took a new event")
}
if _, dropped := log.snapshot(); !slices.Contains(dropped, "x4") {
t.Fatalf("dropped = %v, the refused event's ticket was left held", dropped)
}
}
func TestADeclarationQueuesBehindTheRepoItNames(t *testing.T) {
repo := syntax.DID("did:plc:repo")
record, _ := json.Marshal(org_tangled.RepoDeclaration{Repo: repo.String()})
declaration := &jetstream.Event{Did: "did:plc:owner", Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
Operation: jetstream.CommitOperationCreate, Collection: org_tangled.RepoDeclarationNSID, RKey: "repo", Record: record,
}}
manifest := &jetstream.Event{Did: repo.String(), Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
Operation: jetstream.CommitOperationUpdate, Collection: org_tangled.RepoManifestNSID, RKey: "self",
}}
if ingestShardOf(ingestShardKey(declaration)) != ingestShardOf(ingestShardKey(manifest)) {
t.Fatal("a declaration and the manifest of the repo it names are on different shards")
}
deleted := &jetstream.Event{Did: "did:plc:owner", Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
Operation: jetstream.CommitOperationDelete, Collection: org_tangled.RepoDeclarationNSID, RKey: "repo",
}}
if got := ingestShardKey(deleted); got != "did:plc:owner" {
t.Fatalf("a deleted declaration has no body, key = %q, want its owner", got)
}
}
func TestAPanickingHandlerSkipsItsEventAndTheShardGoesOn(t *testing.T) {
ctx := t.Context()
var applied appliedLog
sh := newIngestShards(ctx, func(_ context.Context, e *jetstream.Event) error {
if e.Commit.RKey == "poison" {
var m map[string]int
m["boom"] = 1
}
applied.add(e.Commit.RKey)
return nil
})
var log ticketLog
dispatchAll(t, ctx, sh, &log, "did:plc:x", "poison", "after")
eventually(t, 5*time.Second, "the event behind the poisoned one to apply", func() bool { return applied.has("after") })
done, dropped := log.snapshot()
if !ticketDone(&log, "poison") || len(dropped) != 0 {
t.Fatalf("done = %v, dropped = %v, want the poisoned event finished so the cursor can pass it", done, dropped)
}
}
// an unfinished ticket stalls shutdown for jetstream's 30s grace
func TestNoTicketIsLeftHeldWhenShutdownRacesABlockedDispatch(t *testing.T) {
for round := range 100 {
ctx, cancel := context.WithCancel(context.Background())
block := make(chan struct{})
sh := newIngestShards(ctx, func(ctx context.Context, e *jetstream.Event) error {
select {
case <-block:
case <-ctx.Done():
}
return nil
})
var log ticketLog
var senders sync.WaitGroup
const events = ingestShardQueue + 40
for i := range events {
name := fmt.Sprintf("r%d-e%d", round, i)
senders.Go(func() { _ = sh.dispatch(ctx, shardEvent("did:plc:x", name), namedTicket{name, &log}) })
}
time.Sleep(time.Millisecond)
cancel()
senders.Wait()
sh.wait()
close(block)
done, dropped := log.snapshot()
if got := len(done) + len(dropped); got != events {
t.Fatalf("round %d: %d of %d tickets were Done or Dropped, the rest stay held", round, got, events)
}
}
}
func TestAnEventThatFailsOnTheDbIsTriedAgainBeforeItIsGivenUp(t *testing.T) {
defer func(old time.Duration) { ingestRetryDelay = old }(ingestRetryDelay)
ingestRetryDelay = time.Millisecond
ctx := t.Context()
var calls atomic.Int32
sh := newIngestShards(ctx, func(context.Context, *jetstream.Event) error {
if calls.Add(1) < 3 {
return errors.New("database is locked")
}
return nil
})
var log ticketLog
dispatchAll(t, ctx, sh, &log, "did:plc:x", "busy")
eventually(t, 5*time.Second, "the event to be applied after the db recovered", func() bool { return ticketDone(&log, "busy") })
if got := calls.Load(); got != 3 {
t.Fatalf("apply ran %d times, want 3 (two failures and the one that worked)", got)
}
}
func TestAnEventThatKeepsFailingLeavesTheCursorBehindItAndTheShardGoesOn(t *testing.T) {
defer func(old time.Duration) { ingestRetryDelay = old }(ingestRetryDelay)
ingestRetryDelay = time.Millisecond
ctx := t.Context()
var applied appliedLog
sh := newIngestShards(ctx, func(_ context.Context, e *jetstream.Event) error {
if e.Commit.RKey == "broken" {
return errors.New("disk I/O error")
}
applied.add(e.Commit.RKey)
return nil
})
var log ticketLog
dispatchAll(t, ctx, sh, &log, "did:plc:x", "broken", "after")
eventually(t, 5*time.Second, "the event behind the broken one to apply", func() bool { return applied.has("after") })
done, dropped := log.snapshot()
if !slices.Equal(dropped, []string{"broken"}) || ticketDone(&log, "broken") {
t.Fatalf("done = %v, dropped = %v, want the failed event dropped so the cursor can't pass it", done, dropped)
}
}