+1

go.mod

Reviewed

··· 47 47 github.com/gorilla/feeds v1.2.0 48 48 github.com/gorilla/sessions v1.4.0 49 49 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 50 50 + github.com/hashicorp/go-version v1.8.0 50 51 github.com/hashicorp/golang-lru/v2 v2.0.7 51 52 github.com/hiddeco/sshsig v0.2.0 52 53 github.com/hpcloud/tail v1.0.0

+2

go.sum

Reviewed

··· 445 445 github.com/hashicorp/go-secure-stdlib/strutil v0.1.2/go.mod h1:Gou2R9+il93BqX25LAKCLuM+y9U2T4hlwvT1yprcna4= 446 446 github.com/hashicorp/go-sockaddr v1.0.7 h1:G+pTkSO01HpR5qCxg7lxfsFEZaG+C0VssTy/9dbT+Fw= 447 447 github.com/hashicorp/go-sockaddr v1.0.7/go.mod h1:FZQbEYa1pxkQ7WLpyXJ6cbjpT8q0YgQaK/JakXqGyWw= 448 448 + github.com/hashicorp/go-version v1.8.0 h1:KAkNb1HAiZd1ukkxDFGmokVZe1Xy9HG6NUp+bPle2i4= 449 449 + github.com/hashicorp/go-version v1.8.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= 448 450 github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= 449 451 github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= 450 452 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=

+3

nix/gomod2nix.toml

Reviewed

··· 479 479 [mod."github.com/hashicorp/go-sockaddr"] 480 480 version = "v1.0.7" 481 481 hash = "sha256-p6eDOrGzN1jMmT/F/f/VJMq0cKNFhUcEuVVwTE6vSrs=" 482 482 + [mod."github.com/hashicorp/go-version"] 483 483 + version = "v1.8.0" 484 484 + hash = "sha256-KXtqERmYrWdpqPCViWcHbe6jnuH7k16bvBIcuJuevj8=" 482 485 [mod."github.com/hashicorp/golang-lru"] 483 486 version = "v1.0.2" 484 487 hash = "sha256-yy+5botc6T5wXgOe2mfNXJP3wr+MkVlUZ2JBkmmrA48="

+8

nix/modules/spindle.nix

Reviewed

··· 32 32 description = "Path to the database file"; 33 33 }; 34 34 35 35 + repoDir = mkOption { 36 36 + type = types.path; 37 37 + default = "/var/lib/spindle/repos"; 38 38 + description = "Path where synced git repositories live"; 39 39 + }; 40 40 + 35 41 hostname = mkOption { 36 42 type = types.str; 37 43 example = "my.spindle.com"; ··· 301 307 302 308 config = let 303 309 deps = [ 310 310 + pkgs.git 304 311 pkgs.qemu 305 312 pkgs.e2fsprogs 306 313 pkgs.slirp4netns ··· 341 348 Environment = [ 342 349 "SPINDLE_SERVER_LISTEN_ADDR=${cfg.server.listenAddr}" 343 350 "SPINDLE_SERVER_DB_PATH=${cfg.server.dbPath}" 351 351 + "SPINDLE_SERVER_REPO_DIR=${cfg.server.repoDir}" 344 352 "SPINDLE_SERVER_HOSTNAME=${cfg.server.hostname}" 345 353 "SPINDLE_SERVER_PLC_URL=${cfg.server.plcUrl}" 346 354 "SPINDLE_SERVER_JETSTREAM_ENDPOINT=${cfg.server.jetstreamEndpoint}"

+1

spindle/config/config.go

Reviewed

··· 12 12 type Server struct { 13 13 ListenAddr string `env:"LISTEN_ADDR, default=0.0.0.0:6555"` 14 14 DBPath string `env:"DB_PATH, default=spindle.db"` 15 15 + RepoDir string `env:"REPO_DIR, default=repos"` 15 16 Hostname string `env:"HOSTNAME, required"` 16 17 JetstreamEndpoint string `env:"JETSTREAM_ENDPOINT, default=wss://jetstream1.us-west.bsky.network/subscribe"` 17 18 Tap Tap `env:",prefix=TAP_"`

+105

spindle/git/git.go

Reviewed

··· 1 1 + package git 2 2 + 3 3 + import ( 4 4 + "bytes" 5 5 + "context" 6 6 + "fmt" 7 7 + "os" 8 8 + "os/exec" 9 9 + "strings" 10 10 + "sync" 11 11 + 12 12 + "github.com/hashicorp/go-version" 13 13 + ) 14 14 + 15 15 + // repoLocks serializes git operations per repo directory. Concurrent triggers 16 16 + // on the same repo (a push landing while a manual run is dispatched, two "Run 17 17 + // CI" clicks, etc.) resolve to the same path with different revisions; running 18 18 + // clone/fetch/checkout there in parallel collides on .git/index.lock and can 19 19 + // corrupt the dir. Locking is keyed by path so unrelated repos don't serialize. 20 20 + var repoLocks keyedMutex 21 21 + 22 22 + type keyedMutex struct { 23 23 + mu sync.Mutex 24 24 + m map[string]*sync.Mutex 25 25 + } 26 26 + 27 27 + // lock acquires the mutex for key and returns its unlock func. 28 28 + func (k *keyedMutex) lock(key string) func() { 29 29 + k.mu.Lock() 30 30 + if k.m == nil { 31 31 + k.m = make(map[string]*sync.Mutex) 32 32 + } 33 33 + mu, ok := k.m[key] 34 34 + if !ok { 35 35 + mu = &sync.Mutex{} 36 36 + k.m[key] = mu 37 37 + } 38 38 + k.mu.Unlock() 39 39 + 40 40 + mu.Lock() 41 41 + return mu.Unlock 42 42 + } 43 43 + 44 44 + func Version() (*version.Version, error) { 45 45 + var buf bytes.Buffer 46 46 + cmd := exec.Command("git", "version") 47 47 + cmd.Stdout = &buf 48 48 + cmd.Stderr = os.Stderr 49 49 + err := cmd.Run() 50 50 + if err != nil { 51 51 + return nil, err 52 52 + } 53 53 + fields := strings.Fields(buf.String()) 54 54 + if len(fields) < 3 { 55 55 + return nil, fmt.Errorf("invalid git version: %s", buf.String()) 56 56 + } 57 57 + 58 58 + // version string is like: "git version 2.29.3" or "git version 2.29.3.windows.1" 59 59 + versionString := fields[2] 60 60 + if pos := strings.Index(versionString, "windows"); pos >= 1 { 61 61 + versionString = versionString[:pos-1] 62 62 + } 63 63 + return version.NewVersion(versionString) 64 64 + } 65 65 + 66 66 + const WorkflowDir = `/.tangled/workflows` 67 67 + 68 68 + func SparseSyncGitRepo(ctx context.Context, cloneUri, path, rev string) error { 69 69 + defer repoLocks.lock(path)() 70 70 + 71 71 + exist, err := isDir(path) 72 72 + if err != nil { 73 73 + return err 74 74 + } 75 75 + if rev == "" { 76 76 + rev = "HEAD" 77 77 + } 78 78 + if !exist { 79 79 + if err := exec.Command("git", "clone", "--no-checkout", "--depth=1", "--filter=tree:0", "--revision="+rev, cloneUri, path).Run(); err != nil { 80 80 + return fmt.Errorf("git clone: %w", err) 81 81 + } 82 82 + if err := exec.Command("git", "-C", path, "sparse-checkout", "set", "--no-cone", WorkflowDir).Run(); err != nil { 83 83 + return fmt.Errorf("git sparse-checkout set: %w", err) 84 84 + } 85 85 + } else { 86 86 + if err := exec.Command("git", "-C", path, "fetch", "--depth=1", "--filter=tree:0", "origin", rev).Run(); err != nil { 87 87 + return fmt.Errorf("git pull: %w", err) 88 88 + } 89 89 + } 90 90 + if err := exec.Command("git", "-C", path, "checkout", rev).Run(); err != nil { 91 91 + return fmt.Errorf("git checkout: %w", err) 92 92 + } 93 93 + return nil 94 94 + } 95 95 + 96 96 + func isDir(path string) (bool, error) { 97 97 + info, err := os.Stat(path) 98 98 + if err == nil && info.IsDir() { 99 99 + return true, nil 100 100 + } 101 101 + if os.IsNotExist(err) { 102 102 + return false, nil 103 103 + } 104 104 + return false, err 105 105 + }

+60

spindle/server.go

Reviewed

··· 8 8 "log/slog" 9 9 "maps" 10 10 "net/http" 11 11 + "path/filepath" 11 12 "sync" 12 13 "time" 13 14 14 15 "github.com/bluesky-social/indigo/atproto/syntax" 15 16 "github.com/go-chi/chi/v5" 17 17 + "github.com/hashicorp/go-version" 16 18 "tangled.org/core/api/tangled" 17 19 "tangled.org/core/eventconsumer" 18 20 "tangled.org/core/eventconsumer/cursor" ··· 28 30 "tangled.org/core/spindle/engines/dummy" 29 31 "tangled.org/core/spindle/engines/microvm" 30 32 "tangled.org/core/spindle/engines/nixery" 33 33 + "tangled.org/core/spindle/git" 31 34 "tangled.org/core/spindle/models" 32 35 "tangled.org/core/spindle/queue" 33 36 "tangled.org/core/spindle/secrets" ··· 328 331 return fmt.Errorf("failed to load config: %w", err) 329 332 } 330 333 334 334 + if err := ensureGitVersion(); err != nil { 335 335 + return fmt.Errorf("ensuring git version: %w", err) 336 336 + } 337 337 + 331 338 d, err := db.Make(ctx, cfg.Server.DBPath) 332 339 if err != nil { 333 340 return fmt.Errorf("failed to setup db: %w", err) ··· 389 396 } 390 397 391 398 func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { 399 399 + l := log.FromContext(ctx).With("handler", "processKnotStream") 400 400 + l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) 392 401 if msg.Nsid == tangled.PipelineNSID { 402 402 + return nil 393 403 tpl := tangled.Pipeline{} 394 404 err := json.Unmarshal(msg.EventJson, &tpl) 395 405 if err != nil { ··· 497 507 } else { 498 508 s.l.Error("failed to enqueue pipeline: queue is full") 499 509 } 510 510 + } else if msg.Nsid == tangled.GitRefUpdateNSID { 511 511 + event := tangled.GitRefUpdate{} 512 512 + if err := json.Unmarshal(msg.EventJson, &event); err != nil { 513 513 + l.Error("error unmarshalling", "err", err) 514 514 + return err 515 515 + } 516 516 + l = l.With("repo", event.Repo, "ref", event.Ref, "newSha", event.NewSha) 517 517 + l.Debug("debug") 518 518 + 519 519 + repoDid := syntax.DID(event.Repo) 520 520 + if _, err := s.db.GetRepoByDid(repoDid); err != nil { 521 521 + return fmt.Errorf("unknown repoDid %s: %w", repoDid, err) 522 522 + } 523 523 + 524 524 + // NOTE: we are blindly trusting the knot that it will return only repos it own 525 525 + repoCloneUri := s.newRepoCloneUrl(src.Key(), syntax.DID(event.Repo)) 526 526 + repoPath := s.newRepoPath(syntax.DID(event.Repo)) 527 527 + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, event.NewSha); err != nil { 528 528 + return fmt.Errorf("sync git repo: %w", err) 529 529 + } 530 530 + l.Info("synced git repo") 531 531 + 532 532 + // TODO: plan the pipeline 500 533 } 501 534 535 535 + return nil 536 536 + } 537 537 + 538 538 + // newRepoPath creates a path to store repository by its did and rkey. 539 539 + // The path format would be: `/data/repos/did:plc:foo/sh.tangled.repo/repo-rkey 540 540 + func (s *Spindle) newRepoPath(repo syntax.DID) string { 541 541 + return filepath.Join(s.cfg.Server.RepoDir, repo.String()) 542 542 + } 543 543 + 544 544 + func (s *Spindle) newRepoCloneUrl(knot string, did syntax.DID) string { 545 545 + scheme := "https://" 546 546 + if s.cfg.Server.Dev { 547 547 + scheme = "http://" 548 548 + } 549 549 + return fmt.Sprintf("%s%s/%s", scheme, knot, did) 550 550 + } 551 551 + 552 552 + const RequiredVersion = "2.49.0" 553 553 + 554 554 + func ensureGitVersion() error { 555 555 + v, err := git.Version() 556 556 + if err != nil { 557 557 + return fmt.Errorf("fetching git version: %w", err) 558 558 + } 559 559 + if v.LessThan(version.Must(version.NewVersion(RequiredVersion))) { 560 560 + return fmt.Errorf("installed git version %q is not supported, Spindle requires git version >= %q", v, RequiredVersion) 561 561 + } 502 562 return nil 503 563 } 504 564

+13 -2

spindle/tapclient.go

Reviewed

··· 16 16 "tangled.org/core/log" 17 17 "tangled.org/core/rbac" 18 18 "tangled.org/core/spindle/db" 19 19 + "tangled.org/core/spindle/git" 19 20 "tangled.org/core/tapc" 20 21 ) 21 22 ··· 122 123 src := eventconsumer.NewKnotSource(record.Knot) 123 124 t.spindle.ks.AddSource(t.spindle.rootCtx, src) 124 125 125 125 - if err := t.spindle.db.AddRepo(db.Repo{ 126 126 + repo := db.Repo{ 126 127 Knot: record.Knot, 127 128 Owner: ownerDid, 128 129 Rkey: rkey, 129 130 RepoDid: repoDid, 130 131 CreatedAt: record.CreatedAt, 131 131 - }); err != nil { 132 132 + } 133 133 + 134 134 + if err := t.spindle.db.AddRepo(repo); err != nil { 132 135 l.Error("failed to add repo row", "err", err) 133 136 return fmt.Errorf("add repo: %w", err) 137 137 + } 138 138 + 139 139 + // setup sparse sync 140 140 + repoCloneUri := t.spindle.newRepoCloneUrl(repo.Knot, repo.RepoDid) 141 141 + repoPath := t.spindle.newRepoPath(repo.RepoDid) 142 142 + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, ""); err != nil { 143 143 + return fmt.Errorf("setting up sparse-clone git repo: %w", err) 134 144 } 135 145 136 146 legacyName := "" ··· 192 202 l.Error("failed to delete repo row", "err", err) 193 203 return fmt.Errorf("delete repo row: %w", err) 194 204 } 205 205 + // TODO: clear sparse-synced git repo 195 206 return nil 196 207 } 197 208