package spindle
import (
"context"
"testing"
"time"
comatproto "github.com/bluesky-social/indigo/api/atproto"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/jetstream"
"tangled.org/core/spindle/secrets"
)
func takedownEvent(did syntax.DID) *jetstream.Event {
status := "takendown"
return &jetstream.Event{
Did: did.String(),
Kind: jetstream.EventKindAccount,
Account: &comatproto.SyncSubscribeRepos_Account{Did: did.String(), Active: false, Status: &status},
}
}
type hangingVault struct{ secrets.Manager }
func (hangingVault) RemoveAllSecrets(ctx context.Context, _ secrets.RepoIdentifier) error {
<-ctx.Done()
return ctx.Err()
}
func TestAnAccountJobStopsWaitingOnAHungWipeAndMovesTheCursorOn(t *testing.T) {
s := newWipeTestSpindle(t)
repoDid, owner := syntax.DID("did:plc:repo1"), syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, repoDid, owner)
s.vault = hangingVault{s.vault}
defer func(old time.Duration) { accountWipeBudget = old }(accountWipeBudget)
accountWipeBudget = 100 * time.Millisecond
var log ticketLog
finished := goErr(func() error {
s.ingestAccountJob(context.Background(), takedownEvent(owner), namedTicket{"account", &log})
return nil
})
waitErr(t, finished, 5*time.Second, "an account job waited on a hung wipe far past its budget")
if done, dropped := log.snapshot(); len(done) != 1 || len(dropped) != 0 {
t.Fatalf("done = %v, dropped = %v, want the ticket finished so the cursor moves on", done, dropped)
}
if exists, err := s.db.WipeExists(repoDid); err != nil || !exists {
t.Fatalf("wipe row exists %v, %v, want it left for the repo jobs to finish", exists, err)
}
}
func TestAnAccountEventWaitsForASlotOnTheReader(t *testing.T) {
env := newEnsureEnv(t)
slots := make(chan struct{}, 1)
slots <- struct{}{}
ctx, cancel := context.WithCancel(context.Background())
var log ticketLog
returned := goErr(func() error {
return env.s.startAccountJob(ctx, takedownEvent("did:plc:nobody"), namedTicket{"account", &log}, slots)
})
select {
case err := <-returned:
t.Fatalf("an account event took a slot that wasn't free, returned %v", err)
case <-time.After(200 * time.Millisecond):
}
cancel()
if err := <-returned; err == nil {
t.Fatal("a cancelled wait for a slot reported success")
}
if done, dropped := log.snapshot(); len(done) != 0 || len(dropped) != 1 {
t.Fatalf("done = %v, dropped = %v, want the ticket dropped so the cursor stays before it", done, dropped)
}
}
func TestAnAccountEventWithAFreeSlotReturnsAtOnceAndFinishesItsTicket(t *testing.T) {
env := newEnsureEnv(t)
slots := make(chan struct{}, 1)
var log ticketLog
if err := env.s.startAccountJob(context.Background(), takedownEvent("did:plc:nobody"), namedTicket{"account", &log}, slots); err != nil {
t.Fatal(err)
}
eventually(t, 5*time.Second, "the account job to finish its ticket", func() bool {
done, _ := log.snapshot()
return len(done) == 1
})
eventually(t, 5*time.Second, "the job to give its slot back", func() bool { return len(slots) == 0 })
}