Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633package worker
import ( "context" cryptorand "crypto/rand" "encoding/base64" "encoding/json" "errors" "fmt" "io" "log/slog" "net/http" "net/url" "os" "path/filepath" "regexp" "strings" "sync" "time"
"tangled.org/core/log" "tangled.org/core/migrator/config" "tangled.org/core/migrator/crypto" "tangled.org/core/migrator/db" "tangled.org/core/migrator/git" migratoroauth "tangled.org/core/migrator/oauth" "tangled.org/core/netutil")
const MaxAttempts = 5
var jobDirPattern = regexp.MustCompile(`^job-[0-9]+$`)
type KnotClient interface { CreateRepo(ctx context.Context, ownerDid, knotDid, rkey, name, sourceURL string) (string, error) // polls with migrator service token; generic reason is sufficient for polling Content(ctx context.Context, knotDid, repoDid string) (content string, err error) // speaks as repository owner to fetch raw import failure reason on terminal states DescribeRepo(ctx context.Context, ownerDid, knotDid, repoDid string) (content string, reason *string, err error) PutRepoRecord(ctx context.Context, ownerDid, rkey, name, description, knotDid, repoDid string) error}
type WorkerPool struct { db *db.DB cfg *config.Config proxy *netutil.SafeConnectProxy cmdRunner git.CommandRunner knot KnotClient logger *slog.Logger concurrency int pollInterval time.Duration wg sync.WaitGroup
githubAPIBase string}
func NewWorkerPool( database *db.DB, cfg *config.Config, proxyServer *netutil.SafeConnectProxy, cmdRunner git.CommandRunner, knotClient KnotClient, logger *slog.Logger,) (*WorkerPool, error) { if cmdRunner == nil { cmdRunner = git.RealCommandRunner{MaxDiskBytes: cfg.MaxDiskBytes} } if knotClient == nil { return nil, errors.New("knot client is required") } return &WorkerPool{ db: database, cfg: cfg, proxy: proxyServer, cmdRunner: cmdRunner, knot: knotClient, logger: log.SubLogger(logger, "worker"), concurrency: cfg.Concurrency, pollInterval: 2 * time.Second, }, nil}
func (p *WorkerPool) Start(ctx context.Context) { if err := p.startupRecovery(ctx); err != nil { p.logger.Error("startup recovery encountered error", "err", err) } p.logger.Info("starting worker pool", "concurrency", p.concurrency) for i := 0; i < p.concurrency; i++ { p.wg.Add(1) go func(workerID int) { defer p.wg.Done() p.workerLoop(ctx, workerID) }(i) }}
func (p *WorkerPool) Wait() { p.wg.Wait() }
func (p *WorkerPool) startupRecovery(ctx context.Context) error { if err := os.MkdirAll(p.cfg.WorkDir, 0700); err != nil { return fmt.Errorf("creating work dir %q: %w", p.cfg.WorkDir, err) } entries, err := os.ReadDir(p.cfg.WorkDir) if err == nil { for _, entry := range entries { if jobDirPattern.MatchString(entry.Name()) { _ = os.RemoveAll(filepath.Join(p.cfg.WorkDir, entry.Name())) } } } resetCount, err := p.db.ResetInflightJobsToQueued(ctx) if err != nil { return fmt.Errorf("resetting inflight jobs: %w", err) } if resetCount > 0 { p.logger.Info("recovered inflight jobs to queued", "count", resetCount) } return nil}
func (p *WorkerPool) workerLoop(ctx context.Context, id int) { for { select { case <-ctx.Done(): return default: } job, batch, err := p.db.ClaimNextQueuedJob(ctx) if err != nil { p.logger.Error("error claiming job", "worker", id, "err", err) if !wait(ctx, time.Second) { return } continue } if job == nil { if !wait(ctx, 500*time.Millisecond) { return } continue } p.executeJob(ctx, job, batch) }}
func wait(ctx context.Context, duration time.Duration) bool { timer := time.NewTimer(duration) defer timer.Stop() select { case <-ctx.Done(): return false case <-timer.C: return true }}
func (p *WorkerPool) executeJob(ctx context.Context, job *db.Job, batch *db.Batch) { jobLogger := p.logger.With("jobId", job.ID, "batchId", batch.ID) jobCtx, cancel := context.WithTimeout(ctx, p.cfg.JobTimeout) defer cancel()
var githubToken string if job.Private { if batch.EncryptedToken == nil || *batch.EncryptedToken == "" { p.terminal(ctx, job, batch, db.StatusAuthorizationRequired, strPtr("missing required GitHub credential for private repository")) return } aad := crypto.ComputeAAD(batch.OwnerDid, batch.ID, batch.RequestID) token, err := crypto.Decrypt(p.cfg.ParsedMasterKey, *batch.EncryptedToken, aad) if err != nil { errStr := "failed to decrypt GitHub credential" jobLogger.Error(errStr, "err", err) p.terminal(ctx, job, batch, db.StatusFailed, &errStr) return } githubToken = token }
capability := "" if job.CapabilityToken != nil { capability = *job.CapabilityToken } if job.RepoDid != "" { if err := p.db.UpdateJobStatus(ctx, job.ID, db.StatusImporting, nil); err != nil { p.handleJobFailure(ctx, job, batch, err, "resuming", githubToken, capability) return } content, err := p.knot.Content(jobCtx, job.KnotDid, job.RepoDid) if err != nil { p.handleJobFailure(ctx, job, batch, err, "resuming", githubToken, capability) return } switch content { case "present": p.recordCompleted(jobCtx, ctx, job, batch, githubToken, capability) return case "failed", "partial": default: if err := p.waitUntilPresent(jobCtx, batch.OwnerDid, job.KnotDid, job.RepoDid); err != nil { p.handleJobFailure(ctx, job, batch, err, "waiting", githubToken, capability) return } p.recordCompleted(jobCtx, ctx, job, batch, githubToken, capability) return } }
// credentials must never reach the knot; private repositories mirror locally source := job.SourceURL if job.Private { jobDir := filepath.Join(p.cfg.WorkDir, fmt.Sprintf("job-%d", job.ID)) mirrorDir := filepath.Join(jobDir, "repo.git") marker := filepath.Join(jobDir, "mirror-ready") if _, err := os.Stat(marker); err != nil { _ = os.RemoveAll(jobDir) if err := os.MkdirAll(jobDir, 0700); err != nil { errStr := fmt.Sprintf("creating scratch dir: %v", err) p.terminal(ctx, job, batch, db.StatusFailed, &errStr) return } if err := p.probeSourceSize(jobCtx, job, githubToken); err != nil { p.handleJobFailure(ctx, job, batch, err, "probing", githubToken, capability) return } if err := p.clonePhase(jobCtx, job, mirrorDir, jobDir, githubToken); err != nil { p.handleJobFailure(ctx, job, batch, err, "cloning", githubToken, capability) return } if err := os.WriteFile(marker, nil, 0600); err != nil { p.handleJobFailure(ctx, job, batch, err, "cloning", githubToken, capability) return } } } else if err := p.probePublicSource(jobCtx, job); err != nil { p.handleJobFailure(ctx, job, batch, err, "probing", githubToken, capability) return }
if err := p.db.UpdateJobStatus(ctx, job.ID, db.StatusImporting, nil); err != nil { p.handleJobFailure(ctx, job, batch, err, "publishing", githubToken, capability) return }
// SetJobCapability requires status importing; keep after status update if job.Private { if capability == "" { newCapability, err := newCapabilityToken() if err != nil { p.handleJobFailure(ctx, job, batch, err, "publishing", githubToken, capability) return } capability = newCapability if err := p.db.SetJobCapability(ctx, job.ID, capability); err != nil { p.handleJobFailure(ctx, job, batch, err, "publishing", githubToken, capability) return } } source = fmt.Sprintf("%s/git/%s/%s.git", strings.TrimSuffix(p.cfg.ServiceURL(), "/"), capability, job.Name) }
if job.RepoDid == "" && p.settleExistingRepo(ctx, jobCtx, job, batch, githubToken, capability) { return }
repoDid, err := p.knot.CreateRepo(jobCtx, batch.OwnerDid, job.KnotDid, job.Name, job.Name, source) if err != nil { if errors.Is(err, migratoroauth.ErrRepoExists) && p.settleConflictRepo(ctx, jobCtx, job, batch, githubToken, capability) { return } p.handleJobFailure(ctx, job, batch, err, "creating", githubToken, capability) return } if err := p.db.SetJobRepoDid(ctx, job.ID, repoDid); err != nil { p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability) return } job.RepoDid = repoDid
if err := p.waitUntilPresent(jobCtx, batch.OwnerDid, job.KnotDid, job.RepoDid); err != nil { p.handleJobFailure(ctx, job, batch, err, "waiting", githubToken, capability) return } p.recordCompleted(jobCtx, ctx, job, batch, githubToken, capability)}
func (p *WorkerPool) adoptableRepoDid(ctx context.Context, ownerDid string, job *db.Job) (string, bool, error) { attempts, err := p.db.EarlierRepoAttemptsFor(ctx, ownerDid, job.KnotDid, job.Name) if err != nil { return "", false, err } for _, attempt := range attempts { if db.SameSource(attempt.SourceURL, job.SourceURL) { return attempt.RepoDid, true, nil } } return "", false, nil}
func (p *WorkerPool) settleExistingRepo(ctx, jobCtx context.Context, job *db.Job, batch *db.Batch, githubToken, capability string) bool { repoDid := job.RepoDid if repoDid == "" { found := false var err error repoDid, found, err = p.adoptableRepoDid(ctx, batch.OwnerDid, job) if err != nil || !found { return false } } content, err := p.knot.Content(jobCtx, job.KnotDid, repoDid) if err != nil { p.handleJobFailure(ctx, job, batch, err, "checking", githubToken, capability) return true } switch content { case "present", "pending", "fetching": if err := p.db.SetJobRepoDid(ctx, job.ID, repoDid); err != nil { p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability) return true } job.RepoDid = repoDid if content != "present" { if err := p.waitUntilPresent(jobCtx, batch.OwnerDid, job.KnotDid, repoDid); err != nil { p.handleJobFailure(ctx, job, batch, err, "waiting", githubToken, capability) return true } } p.recordCompleted(jobCtx, ctx, job, batch, githubToken, capability) return true case "failed", "partial": // failed or partial content cannot be settled without restaging via CreateRepo fallback return false default: msg := fmt.Sprintf("a repository named %q already exists on %s, but it holds no import from this migration", job.Name, job.KnotDid) p.terminal(ctx, job, batch, db.StatusFailed, strPtr(msg)) return true }}
// resolves 409 on repo creation by adopting an earlier attempt or failing on foreign reposfunc (p *WorkerPool) settleConflictRepo(ctx, jobCtx context.Context, job *db.Job, batch *db.Batch, githubToken, capability string) bool { repoDid := job.RepoDid if repoDid == "" { found := false var err error repoDid, found, err = p.adoptableRepoDid(ctx, batch.OwnerDid, job) if err != nil { return false } if !found { msg := fmt.Sprintf("a repository named %q already exists on %s and was not created from %s", job.Name, job.KnotDid, job.SourceURL) p.terminal(ctx, job, batch, db.StatusFailed, strPtr(msg)) return true } } content, err := p.knot.Content(jobCtx, job.KnotDid, repoDid) if err != nil { return false } switch content { case "present", "pending", "fetching", "failed", "partial": if err := p.db.SetJobRepoDid(ctx, job.ID, repoDid); err != nil { p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability) return true } job.RepoDid = repoDid if content != "present" { if err := p.waitUntilPresent(jobCtx, batch.OwnerDid, job.KnotDid, repoDid); err != nil { p.handleJobFailure(ctx, job, batch, err, "waiting", githubToken, capability) return true } } p.recordCompleted(jobCtx, ctx, job, batch, githubToken, capability) return true default: msg := fmt.Sprintf("a repository named %q already exists on %s, but it holds no import from this migration", job.Name, job.KnotDid) p.terminal(ctx, job, batch, db.StatusFailed, strPtr(msg)) return true }}
func (p *WorkerPool) recordCompleted(jobCtx, ctx context.Context, job *db.Job, batch *db.Batch, githubToken, capability string) { if err := p.knot.PutRepoRecord(jobCtx, batch.OwnerDid, job.Name, job.Name, job.Description, job.KnotDid, job.RepoDid); err != nil { p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability) return } p.terminal(ctx, job, batch, db.StatusCompleted, nil)}
func (p *WorkerPool) terminal(ctx context.Context, job *db.Job, batch *db.Batch, status db.JobStatus, errMsg *string) { ctx = context.WithoutCancel(ctx) if err := p.db.UpdateJobStatus(ctx, job.ID, status, errMsg); err != nil { p.logger.Error("marking job terminal", "jobId", job.ID, "status", status, "err", err) } if err := p.db.RevokeJobCapability(ctx, job.ID); err != nil { p.logger.Error("revoking git capability", "jobId", job.ID, "err", err) } if err := os.RemoveAll(filepath.Join(p.cfg.WorkDir, fmt.Sprintf("job-%d", job.ID))); err != nil { p.logger.Error("removing scratch directory", "jobId", job.ID, "err", err) } _, _ = p.db.CheckAndScrubBatchToken(ctx, batch.ID)}
func strPtr(s string) *string { return &s }
func sourceHost(rawURL string) string { u, err := url.Parse(rawURL) if err != nil { return "" } return u.Hostname()}
func (p *WorkerPool) probeSourceSize(ctx context.Context, job *db.Job, githubToken string) error { if p.cfg.MaxPackBytes <= 0 || sourceHost(job.SourceURL) != p.cfg.GitHubHost { return nil } owner, name, ok := githubOwnerRepo(job.SourceURL) if !ok { return nil } base := "https://api." + p.cfg.GitHubHost if p.githubAPIBase != "" { base = p.githubAPIBase } req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/repos/"+owner+"/"+name, nil) if err != nil { return nil } if githubToken != "" { req.Header.Set("Authorization", "Bearer "+githubToken) } resp, err := p.gitHubHTTP().Do(req) if err != nil { return nil } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return nil } var payload struct { Size int64 `json:"size"` // the API reports kilobytes } if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&payload); err != nil || payload.Size <= 0 { return nil } if payload.Size*1024 > p.cfg.MaxPackBytes { return fmt.Errorf("%w: %s/%s reports %d KiB, above the %d byte budget", git.ErrPackLimitExceeded, owner, name, payload.Size, p.cfg.MaxPackBytes) } return nil}
// probe before creation so invalid urls do not leave orphan repos on the knotfunc (p *WorkerPool) probePublicSource(ctx context.Context, job *db.Job) error { if err := p.probeSourceSize(ctx, job, ""); err != nil { return err } proxyAddr := "" if p.proxy != nil { proxyAddr = p.proxy.Addr() } env := git.HardenedGitEnv(proxyAddr, "", p.cfg.WorkDir, "", "") output, err := p.cmdRunner.Run(ctx, "", env, "git", "ls-remote", "--heads", job.SourceURL) if err != nil { return git.ClassifyGitError(output, err) } return nil}
func (p *WorkerPool) gitHubHTTP() *http.Client { transport := http.DefaultTransport.(*http.Transport).Clone() if p.proxy != nil { proxyURL, err := url.Parse("http://" + p.proxy.Addr()) if err != nil { transport.Proxy = nil } else { transport.Proxy = http.ProxyURL(proxyURL) } } return &http.Client{Transport: transport, Timeout: 30 * time.Second}}
func githubOwnerRepo(rawURL string) (string, string, bool) { u, err := url.Parse(rawURL) if err != nil { return "", "", false } parts := strings.Split(strings.Trim(u.Path, "/"), "/") if len(parts) < 2 { return "", "", false } owner, name := parts[0], strings.TrimSuffix(parts[1], ".git") if owner == "" || name == "" { return "", "", false } return owner, name, true}
func newCapabilityToken() (string, error) { raw := make([]byte, 32) if _, err := cryptorand.Read(raw); err != nil { return "", fmt.Errorf("generating git capability: %w", err) } return base64.RawURLEncoding.EncodeToString(raw), nil}
func mirrorPackBytes(mirror string) (int64, error) { entries, err := os.ReadDir(filepath.Join(mirror, "objects", "pack")) if os.IsNotExist(err) { return 0, nil } if err != nil { return 0, err } var total int64 for _, entry := range entries { if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".pack") { continue } info, err := entry.Info() if err != nil { return 0, err } total += info.Size() } return total, nil}
func (p *WorkerPool) waitUntilPresent(ctx context.Context, ownerDid, knotDid, repoDid string) error { var lastErr error for { content, err := p.knot.Content(ctx, knotDid, repoDid) switch { case err != nil: lastErr = err case content == "present": return nil case content == "partial" || content == "failed": if reason := p.importReason(ctx, ownerDid, knotDid, repoDid); reason != nil { return fmt.Errorf("knot import %s: %s", content, *reason) } return fmt.Errorf("knot import %s", content) default: lastErr = nil } if !wait(ctx, p.pollInterval) { if lastErr != nil { return fmt.Errorf("waiting for knot content: %w (last answer: %v)", ctx.Err(), lastErr) } return fmt.Errorf("waiting for knot content: %w", ctx.Err()) } }}
func (p *WorkerPool) importReason(ctx context.Context, ownerDid, knotDid, repoDid string) *string { _, reason, err := p.knot.DescribeRepo(ctx, ownerDid, knotDid, repoDid) if err != nil { return nil } return reason}
func (p *WorkerPool) clonePhase(ctx context.Context, job *db.Job, mirrorDir, jobDir, githubToken string) error { var askpassBin, askpassHost, askpassToken string if job.Private { var err error askpassBin, err = git.CreateAskpassScript(jobDir) if err != nil { return fmt.Errorf("creating clone askpass script: %w", err) } defer os.Remove(askpassBin) askpassHost, askpassToken = sourceHost(job.SourceURL), githubToken } proxyAddr := "" if p.proxy != nil { proxyAddr = p.proxy.Addr() } env := git.HardenedGitEnv(proxyAddr, askpassBin, jobDir, askpassHost, askpassToken) runner := p.cmdRunner if real, ok := runner.(git.RealCommandRunner); ok { real.WatchDir = jobDir runner = real } output, err := runner.Run(ctx, jobDir, env, "git", "clone", "--mirror", job.SourceURL, mirrorDir) if err != nil { return git.ClassifyGitError(output, err) } if p.cfg.MaxPackBytes > 0 { packBytes, err := mirrorPackBytes(mirrorDir) if err != nil { return fmt.Errorf("measuring cloned pack: %w", err) } if packBytes > p.cfg.MaxPackBytes { return fmt.Errorf("%w: cloned pack is %d bytes, max %d", git.ErrPackLimitExceeded, packBytes, p.cfg.MaxPackBytes) } } if _, err := runner.LookPath("git-lfs"); err == nil { output, err := runner.Run(ctx, mirrorDir, env, "git", "lfs", "fetch", "--all") if err != nil { p.logger.Error("git lfs fetch failed", "out", git.RedactSecrets(output, githubToken), "err", git.RedactSecrets(err.Error(), githubToken)) return git.ClassifyGitError(output, err) } } return nil}
func (p *WorkerPool) handleJobFailure(ctx context.Context, job *db.Job, batch *db.Batch, err error, phase, githubToken, capability string) { stopping := ctx.Err() != nil safeErr := git.RedactSecrets(err.Error(), githubToken, capability) p.logger.Error("job failed during phase", "jobId", job.ID, "phase", phase, "attempts", job.Attempts, "err", safeErr) ctx = context.WithoutCancel(ctx) // refused grants invalidate the refresh-token family; retrying cannot succeed if errors.Is(err, git.ErrAuthRequired) || errors.Is(err, migratoroauth.ErrGrantRequired) { p.terminal(ctx, job, batch, db.StatusAuthorizationRequired, &safeErr) return } if errors.Is(err, git.ErrDiskLimitExceeded) || errors.Is(err, git.ErrPackLimitExceeded) { p.terminal(ctx, job, batch, db.StatusFailed, &safeErr) return } if stopping { if err := p.db.ReleaseJob(ctx, job.ID, &safeErr); err != nil { p.logger.Error("releasing job after cancellation", "jobId", job.ID, "err", err) } return } if job.Attempts < MaxAttempts { backoff := min(time.Duration(1<<uint(job.Attempts))*time.Second, 30*time.Second) _ = p.db.ScheduleJobRetry(ctx, job.ID, backoff, &safeErr) return } p.terminal(ctx, job, batch, db.StatusFailed, &safeErr)}