//go:build torture
package spindle
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"math"
"net/http"
"net/http/cgi"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/bluesky-social/indigo/atproto/atcrypto"
"github.com/bluesky-social/indigo/atproto/auth"
"github.com/bluesky-social/indigo/atproto/identity"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/sethvargo/go-envconfig"
"tangled.org/core/api/org_tangled"
"tangled.org/core/idresolver"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/knotclient"
"tangled.org/core/spindle/models"
"tangled.org/core/spindle/quota"
"tangled.org/core/tapc"
"tangled.org/core/xrpc/serviceauth"
)
// one directory for every did, knots too, since a knot is did:web on loopback
type tortureDirectory struct {
*identity.MockDirectory
delay atomic.Int64
lookups atomic.Int64
}
func newTortureDirectory() *tortureDirectory {
return &tortureDirectory{MockDirectory: identity.NewMockDirectory()}
}
func (d *tortureDirectory) LookupDID(ctx context.Context, did syntax.DID) (*identity.Identity, error) {
d.lookups.Add(1)
if err := sleepCtx(ctx, time.Duration(d.delay.Load())); err != nil {
return nil, err
}
return d.MockDirectory.LookupDID(ctx, did)
}
func (d *tortureDirectory) Lookup(ctx context.Context, atid syntax.AtIdentifier) (*identity.Identity, error) {
if did, err := atid.AsDID(); err == nil {
return d.LookupDID(ctx, did)
}
return d.MockDirectory.Lookup(ctx, atid)
}
func sleepCtx(ctx context.Context, d time.Duration) error {
if d <= 0 {
return nil
}
t := time.NewTimer(d)
defer t.Stop()
select {
case <-t.C:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
type tortureKnot struct {
*fakeKnot
did syntax.DID
priv atcrypto.PrivateKey
git *cgi.Handler
root string
delay atomic.Int64
gitDelay atomic.Int64
down atomic.Bool
xrpc atomic.Int64
fetches atomic.Int64
inflight atomic.Int64
readsMu sync.Mutex
reads map[string]int64
}
func (k *tortureKnot) readsOf(collection string) int64 {
k.readsMu.Lock()
defer k.readsMu.Unlock()
return k.reads[collection]
}
func (k *tortureKnot) url() string { return k.srv.URL }
func (k *tortureKnot) middleware(records http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
k.inflight.Add(1)
defer k.inflight.Add(-1)
if strings.HasPrefix(r.URL.Path, "/did:") {
k.serveGit(w, r)
return
}
k.xrpc.Add(1)
q := r.URL.Query()
if collection := q.Get("collection"); collection != "" {
k.readsMu.Lock()
k.reads[collection]++
k.readsMu.Unlock()
}
if sleepCtx(r.Context(), time.Duration(k.delay.Load())) != nil {
return
}
if k.down.Load() {
http.Error(w, "boom", http.StatusInternalServerError)
return
}
records.ServeHTTP(w, r)
})
}
func (k *tortureKnot) serveGit(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/info/refs") {
k.fetches.Add(1)
}
if sleepCtx(r.Context(), time.Duration(k.gitDelay.Load())) != nil {
return
}
k.git.ServeHTTP(w, r)
}
type tortureRepo struct {
did syntax.DID
owner syntax.DID
name string
knot *tortureKnot
shas []string
}
type tortureWorld struct {
r *tortureRun
dir string
js *fakeJetstream
hyd *fakeHydrant
ids *tortureDirectory
knots []*tortureKnot
eng *tortureEngine
pdsMu sync.Mutex
pds map[syntax.DID]*tortureKnot
// read at boot, set these before the world starts
maxJobs int
admission config.Admission
qm *quota.Manager
skew atomic.Int64
repoPool, notifySlots int
priorityKnots []string
s *Spindle
sp atomic.Pointer[Spindle]
api *httptest.Server
cancel context.CancelFunc
done chan error
nextRepo atomic.Int64
}
func newTortureWorld(r *tortureRun) *tortureWorld {
t := r.t
t.Setenv("GIT_CONFIG_GLOBAL", os.DevNull)
t.Setenv("GIT_CONFIG_SYSTEM", os.DevNull)
t.Setenv("GIT_AUTHOR_NAME", "torture")
t.Setenv("GIT_AUTHOR_EMAIL", "torture@tangled.org")
t.Setenv("GIT_COMMITTER_NAME", "torture")
t.Setenv("GIT_COMMITTER_EMAIL", "torture@tangled.org")
gitExecPath(t)
w := &tortureWorld{
r: r,
dir: t.TempDir(),
ids: newTortureDirectory(),
pds: map[syntax.DID]*tortureKnot{},
eng: newTortureEngine(),
maxJobs: 8,
admission: config.AdmissionOpen,
}
w.js = newFakeJetstream(t)
w.hyd = &fakeHydrant{}
w.hyd.onTrack = func(did syntax.DID) {
if w.hyd.repoStatus(did) != synced(did).Status {
w.backfill(did)
}
}
w.hyd.onResync = func(did syntax.DID) {
if w.hyd.repoStatus(did) != "" {
w.backfill(did)
}
}
w.ids.Insert(identity.Identity{DID: testOwner})
r.gauge("queued_jobs", func() float64 { return w.rows("jobs") })
r.gauge("pipelines", func() float64 { return w.rows("pipelines") })
r.gauge("repo_retries", func() float64 { return w.rows("repo_retries") })
r.gauge("db_mb", func() float64 {
var total int64
for _, suffix := range []string{"", "-wal", "-shm"} {
if fi, err := os.Stat(w.dbPath() + suffix); err == nil {
total += fi.Size()
}
}
return float64(total) / (1 << 20)
})
r.gauge("engine_running", func() float64 { return float64(w.eng.running.Load()) })
return w
}
func (w *tortureWorld) dbPath() string { return filepath.Join(w.dir, "spindle.db") }
func (w *tortureWorld) spindleDid() string {
return config.Server{Hostname: faultHostname}.Did().String()
}
// nil once stopped
func (w *tortureWorld) spindle() *Spindle { return w.sp.Load() }
// NaN once stopped, since a `<=` ceiling check fails closed on it
func (w *tortureWorld) fromSpindle(read func(s *Spindle) float64) float64 {
s := w.spindle()
if s == nil {
return math.NaN()
}
return read(s)
}
func (w *tortureWorld) countRows(query string) float64 {
return w.fromSpindle(func(s *Spindle) float64 {
var n float64
if err := s.db.QueryRow(query).Scan(&n); err != nil {
return math.NaN()
}
return n
})
}
func (w *tortureWorld) newKeyedIdentity(did syntax.DID) (atcrypto.PrivateKey, identity.Identity) {
priv, err := atcrypto.GeneratePrivateKeyP256()
if err != nil {
w.r.t.Fatal(err)
}
pub, err := priv.PublicKey()
if err != nil {
w.r.t.Fatal(err)
}
return priv, identity.Identity{
DID: did,
Keys: map[string]identity.VerificationMethod{"atproto": {Type: "Multikey", PublicKeyMultibase: pub.Multibase()}},
}
}
func (w *tortureWorld) probeTable() {
if _, err := w.spindle().db.Exec(`create table if not exists torture_probe (n integer)`); err != nil {
w.r.t.Fatal(err)
}
}
// inserts on a timer, because a starved writer shows up as latency
func (w *tortureWorld) probeWrites(every time.Duration, lat *latencies, errs *atomic.Int64) (stop func()) {
done := make(chan struct{})
var wg sync.WaitGroup
wg.Go(func() {
tick := time.NewTicker(every)
defer tick.Stop()
for i := 0; ; i++ {
select {
case <-done:
return
case <-tick.C:
}
start := time.Now()
if _, err := w.spindle().db.Exec(`insert into torture_probe (n) values (?)`, i); err != nil {
errs.Add(1)
continue
}
lat.since(start)
}
})
return func() {
close(done)
wg.Wait()
}
}
func (w *tortureWorld) addKnots(n int) []*tortureKnot {
ks := make([]*tortureKnot, n)
for i := range ks {
ks[i] = w.addKnot()
}
return ks
}
func (w *tortureWorld) rows(table string) float64 {
return w.countRows("select count(*) from " + table)
}
func (w *tortureWorld) addKnot() *tortureKnot {
t := w.r.t
k := &tortureKnot{reads: map[string]int64{}}
k.fakeKnot = newFakeKnotBehind(t, k.middleware)
k.root = filepath.Join(w.dir, fmt.Sprintf("knot%d", len(w.knots)))
if err := os.MkdirAll(k.root, 0o755); err != nil {
t.Fatal(err)
}
k.git = &cgi.Handler{
Path: gitExecPath(t),
Root: "/",
Env: []string{"GIT_PROJECT_ROOT=" + k.root, "GIT_HTTP_EXPORT_ALL=1"},
InheritEnv: []string{"PATH", "HOME", "GIT_CONFIG_GLOBAL", "GIT_CONFIG_SYSTEM"},
}
k.did = serviceauth.DidWeb(k.host())
var id identity.Identity
k.priv, id = w.newKeyedIdentity(k.did)
w.ids.Insert(id)
w.knots = append(w.knots, k)
return k
}
func tortureWorkflow(steps ...string) string {
var b strings.Builder
b.WriteString("engine: torture\nwhen:\n - event: push\n branch: main\nsteps:\n")
for i, step := range steps {
fmt.Fprintf(&b, " - name: s%d\n command: %q\n", i, step)
}
return b.String()
}
// no commits skips git, the repo is only records for membership load
func (w *tortureWorld) addRepo(k *tortureKnot, owner syntax.DID, commits int, workflows map[string]string) *tortureRepo {
repo, err := w.tryAddRepo(k, owner, commits, workflows)
if err != nil {
w.r.t.Fatal(err)
}
return repo
}
// no t.Fatal here, addRepos calls this off the test goroutine
func (w *tortureWorld) tryAddRepo(k *tortureKnot, owner syntax.DID, commits int, workflows map[string]string) (*tortureRepo, error) {
n := w.nextRepo.Add(1)
repo := &tortureRepo{did: syntax.DID(fmt.Sprintf("did:plc:torture%06d", n)), owner: owner, name: fmt.Sprintf("repo%d", n), knot: k}
if commits > 0 {
if err := w.addHistory(repo, commits, workflows); err != nil {
return nil, err
}
}
w.ids.Insert(identity.Identity{DID: repo.did, Services: map[string]identity.ServiceEndpoint{
"tangled_knot": {Type: "TangledKnot", URL: k.url()},
}})
k.put(repo.did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, manifestRecord(owner, repo.name, w.spindleDid()))
k.put(repo.did, org_tangled.GitRefsNSID, knotclient.RefsRkey, org_tangled.GitRefs{LexiconTypeID: org_tangled.GitRefsNSID, DefaultBranch: "main"})
w.pdsOf(owner, k).declare(owner, repo)
return repo, nil
}
func (w *tortureWorld) pdsOf(owner syntax.DID, k *tortureKnot) *tortureKnot {
w.pdsMu.Lock()
defer w.pdsMu.Unlock()
if pds, ok := w.pds[owner]; ok {
return pds
}
w.pds[owner] = k
w.ids.Insert(identity.Identity{DID: owner, Services: map[string]identity.ServiceEndpoint{
"atproto_pds": {Type: "AtprotoPersonalDataServer", URL: k.url()},
}})
return k
}
func (w *tortureWorld) addHistory(repo *tortureRepo, commits int, workflows map[string]string) error {
k := repo.knot
bare := filepath.Join(k.root, repo.did.String())
gitCmd := func(stdin io.Reader, args ...string) error {
cmd := exec.Command("git", append([]string{"-C", bare}, args...)...)
cmd.Stdin = stdin
if out, err := cmd.CombinedOutput(); err != nil {
return fmt.Errorf("git %v: %w\n%s", args, err, out)
}
return nil
}
if err := os.MkdirAll(bare, 0o755); err != nil {
return err
}
for _, args := range [][]string{
{"init", "-q", "--bare", "-b", "main"},
{"config", "uploadpack.allowFilter", "true"},
{"config", "uploadpack.allowAnySHA1InWant", "true"},
} {
if err := gitCmd(nil, args...); err != nil {
return err
}
}
var stream bytes.Buffer
names := make([]string, 0, len(workflows))
for name := range workflows {
names = append(names, name)
}
sort.Strings(names)
blob := func(path, contents string) {
fmt.Fprintf(&stream, "M 644 inline %s\ndata %d\n%s\n", path, len(contents), contents)
}
for i := 1; i <= commits; i++ {
fmt.Fprintf(&stream, "commit refs/heads/c%d\nmark :%d\ncommitter t <t@t> %d +0000\ndata 2\nc\n", i, i, 1700000000+i)
if i > 1 {
fmt.Fprintf(&stream, "from :%d\n", i-1)
blob("n", repo.did.String()+" "+strconv.Itoa(i))
} else {
blob("n", repo.did.String()+" 1")
for _, name := range names {
blob(".tangled/workflows/"+name, workflows[name])
}
}
}
marks := filepath.Join(bare, "torture-marks")
if err := gitCmd(&stream, "fast-import", "--quiet", "--export-marks="+marks); err != nil {
return err
}
raw, err := os.ReadFile(marks)
if err != nil {
return err
}
repo.shas = make([]string, commits)
for _, line := range strings.Split(strings.TrimSpace(string(raw)), "\n") {
mark, sha, _ := strings.Cut(line, " ")
i, _ := strconv.Atoi(strings.TrimPrefix(mark, ":"))
repo.shas[i-1] = sha
}
return gitCmd(nil, "update-ref", "refs/heads/main", repo.shas[commits-1])
}
func (k *tortureKnot) declare(owner syntax.DID, repo *tortureRepo) {
k.put(owner, org_tangled.RepoDeclarationNSID, strings.ToLower(repo.name), org_tangled.RepoDeclaration{LexiconTypeID: org_tangled.RepoDeclarationNSID, Repo: repo.did.String()})
}
// repos go in parallel because git is most of the setup time
func (w *tortureWorld) addRepos(n int, pick func(i int) (*tortureKnot, syntax.DID), commits int, workflows map[string]string) []*tortureRepo {
repos := make([]*tortureRepo, n)
var wg sync.WaitGroup
var firstErr atomic.Pointer[error]
next := atomic.Int64{}
for range 8 {
wg.Go(func() {
for {
i := int(next.Add(1)) - 1
if i >= n {
return
}
k, owner := pick(i)
repo, err := w.tryAddRepo(k, owner, commits, workflows)
if err != nil {
firstErr.CompareAndSwap(nil, &err)
return
}
repos[i] = repo
}
})
}
wg.Wait()
if err := firstErr.Load(); err != nil {
w.r.t.Fatal(*err)
}
return repos
}
func (w *tortureWorld) emitManifest(repo *tortureRepo) int64 {
return w.js.manifest(repo.did, repo.knot.record(repo.did, org_tangled.RepoManifestNSID, knotclient.ManifestRkey))
}
func (w *tortureWorld) emitDeclaration(repo *tortureRepo) int64 {
return w.js.declaration(repo.owner, repo.name, repo.did)
}
func (w *tortureWorld) emitCollaborator(repo *tortureRepo, who syntax.DID, op string) int64 {
return w.js.collaborator(repo.did, who, op)
}
// production learns repos from jetstream and seed does too
func (w *tortureWorld) seed(repos []*tortureRepo, within time.Duration) {
start := time.Now()
var last int64
for _, repo := range repos {
w.emitDeclaration(repo)
last = w.emitManifest(repo)
}
if !w.r.waitFor(within, func() bool { return w.servedCount(repos) == len(repos) && w.s.jc.LastSeenUs() > last }) {
w.r.t.Fatalf("only %d of %d repos served after the jetstream seed", w.servedCount(repos), len(repos))
}
w.r.metric("seed_s", time.Since(start).Seconds())
}
func (w *tortureWorld) servedCount(repos []*tortureRepo) int {
n := 0
for _, repo := range repos {
if w.served(repo) {
n++
}
}
return n
}
func (w *tortureWorld) served(repo *tortureRepo) bool {
s := w.spindle()
if s == nil {
return false
}
rows, err := s.db.ReposByDid(repo.did)
return err == nil && len(rows) > 0
}
func (w *tortureWorld) consumed(us int64, within time.Duration) bool {
return w.r.waitFor(within, func() bool {
s := w.spindle()
return s != nil && s.jc.LastSeenUs() > us
})
}
func (w *tortureWorld) backfill(did syntax.DID) {
var events []tapc.RecordEventData
for _, k := range w.knots {
events = append(events, k.recordsOf(did)...)
}
w.hyd.backfill(did, events)
}
func (w *tortureWorld) boot() {
t := w.r.t
ctx := context.Background()
d, err := db.Make(ctx, w.dbPath())
if err != nil {
t.Fatal(err)
}
cfg := &config.Config{}
if err := envconfig.ProcessWith(ctx, &envconfig.Config{
Target: cfg,
Lookuper: envconfig.MapLookuper(map[string]string{"SPINDLE_SERVER_HOSTNAME": faultHostname}),
}); err != nil {
t.Fatal(err)
}
cfg.Server.ListenAddr = "127.0.0.1:0"
cfg.Server.MetricsListenAddr = "127.0.0.1:0"
cfg.Server.DBPath = w.dbPath()
cfg.Server.RepoDir = filepath.Join(w.dir, "repos")
cfg.Server.LogDir = filepath.Join(w.dir, "logs")
cfg.Server.JetstreamEndpoint = w.js.url()
cfg.Server.PlcUrl = "http://127.0.0.1:1"
cfg.Server.Hydrant = config.Hydrant{Dir: shortTempDir(t)}
cfg.Server.Dev = true
cfg.Server.Owner = testOwner.String()
cfg.Server.Admission = w.admission
cfg.Server.MaxJobCount = w.maxJobs
if w.repoPool > 0 {
cfg.Server.RepoJobWorkers = w.repoPool
}
if w.notifySlots > 0 {
cfg.Server.NotifySlots = w.notifySlots
}
if len(w.priorityKnots) > 0 {
cfg.Server.PriorityKnots = w.priorityKnots
}
cfg.ArtifactStores.Disk.Dir = filepath.Join(w.dir, "artifacts")
for _, dir := range []string{cfg.Server.RepoDir, cfg.Server.LogDir, cfg.ArtifactStores.Disk.Dir} {
if err := os.MkdirAll(dir, 0o755); err != nil {
t.Fatal(err)
}
}
connects := w.js.connects()
s, err := New(ctx, cfg, d, map[string]models.Engine{"torture": w.eng}, w.qm)
if err != nil {
t.Fatalf("New: %v", err)
}
dir := identity.NewCacheDirectory(w.ids, 250_000, 24*time.Hour, 2*time.Minute, 5*time.Minute)
s.res = idresolver.NewMockResolver(dir)
s.knots = knotclient.New(dir, knotclient.HTTPClient(true), true)
s.backfill.start = w.hyd.start
s.retries.now = func() time.Time { return time.Now().Add(time.Duration(w.skew.Load())) }
w.s = s
w.sp.Store(s)
w.api = httptest.NewServer(s.Router())
runCtx, cancel := context.WithCancel(ctx)
w.cancel = cancel
w.done = make(chan error, 1)
go func() { w.done <- s.Start(runCtx) }()
t.Cleanup(w.stop)
if !w.r.waitFor(10*time.Second, func() bool { return w.js.connects() > connects }) {
t.Fatal("jetstream never connected")
}
}
// the db stays on disk for the next boot to reopen
func (w *tortureWorld) stop() {
if w.cancel == nil {
return
}
w.api.Close()
w.cancel()
w.cancel = nil
select {
case err := <-w.done:
if err != nil {
w.r.t.Errorf("Start: %v", err)
}
case <-time.After(60 * time.Second):
w.r.check("clean shutdown", false, "spindle did not stop within 60s")
}
w.sp.Store(nil)
w.s.db.Close()
w.s = nil
}
type notifyReply struct {
status int
body string
elapsed time.Duration
}
func (w *tortureWorld) notifyBody(repo *tortureRepo, ref, oldSha, newSha string) []byte {
actor := repo.owner.String()
body, err := json.Marshal(org_tangled.EventNotifyEvent_Input{
Repo: repo.did.String(),
Event: &org_tangled.EventNotifyEvent_Input_Event{EventPush: &org_tangled.EventPush{
Repo: repo.did.String(), Ref: ref, OldSha: oldSha, NewSha: newSha, Actor: &actor,
}},
})
if err != nil {
panic(err)
}
return body
}
func (w *tortureWorld) sign(did syntax.DID, priv atcrypto.PrivateKey, nsid string) string {
lxm := syntax.NSID(nsid)
token, err := auth.SignServiceAuth(did, w.spindleDid(), time.Minute, &lxm, priv)
if err != nil {
panic(err)
}
return token
}
func (w *tortureWorld) token(k *tortureKnot) string {
return w.sign(k.did, k.priv, org_tangled.EventNotifyEventNSID)
}
func (w *tortureWorld) call(ctx context.Context, client *http.Client, method, nsid, query, token string, body []byte) notifyReply {
start := time.Now()
target := w.api.URL + "/xrpc/" + nsid
if query != "" {
target += "?" + query
}
req, err := http.NewRequestWithContext(ctx, method, target, bytes.NewReader(body))
if err != nil {
panic(err)
}
req.Header.Set("Content-Type", "application/json")
if token != "" {
req.Header.Set("Authorization", "Bearer "+token)
}
resp, err := client.Do(req)
if err != nil {
return notifyReply{status: -1, body: err.Error(), elapsed: time.Since(start)}
}
defer resp.Body.Close()
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<20))
return notifyReply{status: resp.StatusCode, body: string(raw), elapsed: time.Since(start)}
}
func (w *tortureWorld) post(ctx context.Context, client *http.Client, token string, body []byte) notifyReply {
return w.call(ctx, client, http.MethodPost, org_tangled.EventNotifyEventNSID, "", token, body)
}
func statusKind(status int) string {
if status < 0 {
return "transport"
}
return "http_" + strconv.Itoa(status)
}