Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
12 kB · 446 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447package knotmirror
import ( "bytes" "context" "database/sql" "encoding/json" "errors" "fmt" "io" "log/slog" "math/rand" "net/http" "net/url" "strings" "sync" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/knotmirror/config" "tangled.org/core/knotmirror/db" "tangled.org/core/knotmirror/knotstream" "tangled.org/core/knotmirror/models" "tangled.org/core/log")
type Resyncer struct { logger *slog.Logger db *sql.DB gitm GitMirrorManager cfg *config.Config indexer *knotstream.ParallelScheduler
claimJobMu sync.Mutex
runningJobs map[syntax.DID]context.CancelFunc runningJobsMu sync.Mutex
repoFetchTimeout time.Duration manualResyncTimeout time.Duration parallelism int
knotBackoff map[string]time.Time knotBackoffMu sync.RWMutex
httpClient *http.Client}
func NewResyncer(l *slog.Logger, db *sql.DB, gitm GitMirrorManager, indexer *knotstream.ParallelScheduler, cfg *config.Config) *Resyncer { return &Resyncer{ logger: log.SubLogger(l, "resyncer"), db: db, gitm: gitm, cfg: cfg, indexer: indexer,
runningJobs: make(map[syntax.DID]context.CancelFunc),
repoFetchTimeout: cfg.GitRepoFetchTimeout, manualResyncTimeout: 30 * time.Minute, parallelism: cfg.ResyncParallelism,
knotBackoff: make(map[string]time.Time),
httpClient: &http.Client{Timeout: 30 * time.Second}, }}
func (r *Resyncer) Start(ctx context.Context) { for i := 0; i < r.parallelism; i++ { go r.runResyncWorker(ctx, i) }}
func (r *Resyncer) runResyncWorker(ctx context.Context, workerID int) { l := r.logger.With("worker", workerID) for { select { case <-ctx.Done(): l.Info("resync worker shutting down", "error", ctx.Err()) return default: } repoDid, found, err := r.claimResyncJob(ctx) if err != nil { l.Error("failed to claim resync job", "error", err) time.Sleep(time.Second) continue } if !found { time.Sleep(time.Second) continue } l.Info("processing resync", "did", repoDid) if err := r.resyncRepo(ctx, repoDid); err != nil { l.Error("resync failed", "did", repoDid, "error", err) } }}
func (r *Resyncer) registerRunning(repo syntax.DID, cancel context.CancelFunc) { r.runningJobsMu.Lock() defer r.runningJobsMu.Unlock()
if _, exists := r.runningJobs[repo]; exists { return } r.runningJobs[repo] = cancel}
func (r *Resyncer) unregisterRunning(repo syntax.DID) { r.runningJobsMu.Lock() defer r.runningJobsMu.Unlock()
delete(r.runningJobs, repo)}
func (r *Resyncer) CancelResyncJob(repo syntax.DID) { r.runningJobsMu.Lock() defer r.runningJobsMu.Unlock()
cancel, ok := r.runningJobs[repo] if !ok { return } delete(r.runningJobs, repo) cancel()}
// TriggerResyncJob manually triggers the resync jobfunc (r *Resyncer) TriggerResyncJob(ctx context.Context, repoDid syntax.DID) error { repo, err := db.GetRepoByRepoDid(ctx, r.db, repoDid) if err != nil { return fmt.Errorf("failed to get repo: %w", err) } if repo == nil { return fmt.Errorf("repo not found: %s", repoDid) }
if repo.State == models.RepoStateResyncing { return fmt.Errorf("repo already resyncing") }
repo.State = models.RepoStatePending repo.RetryAfter = -1 // resyncer will prioritize this
if err := db.UpsertRepo(ctx, r.db, repo); err != nil { return fmt.Errorf("updating repo state to pending %w", err) } return nil}
func (r *Resyncer) claimResyncJob(ctx context.Context) (syntax.DID, bool, error) { // use mutex to prevent duplicated jobs r.claimJobMu.Lock() defer r.claimJobMu.Unlock()
var repoDid syntax.DID now := time.Now().Unix() if err := r.db.QueryRowContext(ctx, `update repos set state = $1 where repo_did = ( select repo_did from repos where state in ($2, $3, $4) and (retry_after = -1 or retry_after = 0 or retry_after < $5) order by (retry_after = -1) desc, (retry_after = 0) desc, retry_after limit 1 ) returning repo_did `, models.RepoStateResyncing, models.RepoStatePending, models.RepoStateDesynchronized, models.RepoStateError, now, ).Scan(&repoDid); err != nil { if errors.Is(err, sql.ErrNoRows) { return "", false, nil } return "", false, err }
return repoDid, true, nil}
func (r *Resyncer) resyncRepo(ctx context.Context, repoDid syntax.DID) error { // ctx, span := tracer.Start(ctx, "resyncRepo") // span.SetAttributes(attribute.String("aturi", repoAt)) // defer span.End()
resyncsStarted.Inc() startTime := time.Now()
jobCtx, cancel := context.WithCancel(ctx) r.registerRunning(repoDid, cancel) defer r.unregisterRunning(repoDid)
success, err := r.doResync(jobCtx, repoDid) if !success { resyncsFailed.Inc() resyncDuration.Observe(time.Since(startTime).Seconds()) return r.handleResyncFailure(ctx, repoDid, err) }
resyncsCompleted.Inc() resyncDuration.Observe(time.Since(startTime).Seconds()) return nil}
func (r *Resyncer) doResync(ctx context.Context, repoDid syntax.DID) (bool, error) { // ctx, span := tracer.Start(ctx, "doResync") // span.SetAttributes(attribute.String("aturi", repoAt)) // defer span.End()
repo, err := db.GetRepoByRepoDid(ctx, r.db, repoDid) if err != nil { return false, fmt.Errorf("failed to get repo: %w", err) } if repo == nil { // untracked repo, skip return false, nil }
r.knotBackoffMu.RLock() backoffUntil, inBackoff := r.knotBackoff[repo.KnotDomain] r.knotBackoffMu.RUnlock() if inBackoff && time.Now().Before(backoffUntil) { return false, nil }
// HACK: check knot reachability with short timeout before running actual fetch. // This is crucial as git-cli doesn't support http connection timeout. // `http.lowSpeedTime` is only applied _after_ the connection. format, err := r.checkKnot(ctx, repo) if err != nil { if isRateLimitError(err) { r.knotBackoffMu.Lock() r.knotBackoff[repo.KnotDomain] = time.Now().Add(10 * time.Second) r.knotBackoffMu.Unlock() return false, nil } // TODO: suspend repo on 404. KnotStream updates will change the repo state back online return false, fmt.Errorf("knot unreachable: %w", err) }
if format == models.ObjectFormatSHA256 { return r.suspendUnsupported(ctx, repo) }
timeout := r.repoFetchTimeout if repo.RetryAfter == -1 { timeout = r.manualResyncTimeout } fetchCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel()
if err := r.gitm.Sync(fetchCtx, repo); err != nil { return false, err }
// request index to zoekt server // NOTE: indexing after full git resync is bad design. We are doing _after_ the sync because knotstream event doesn't include repository refs. // NOTE: and zoekt indexer should directly subscribe to the knot. remove this when we have knotrelay. if r.cfg.Search.ZoektUrl != "" { go func() { idxCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() defaultBranch, err := r.gitm.DefaultBranch(idxCtx, repo) if err != nil { r.logger.Warn("resolving default branch for indexing failed", "did", repo.RepoDid, "error", err) return } if err := r.requestIndex(idxCtx, repo.RepoDid, []branch{defaultBranch}); err != nil { r.logger.Warn("requesting zoekt index failed", "did", repo.RepoDid, "err", err) } }() }
// queue repo_stats_update job r.indexer.AddTask(context.TODO(), &knotstream.Task{Key: repo.RepoDid.String()})
// repo.GitRev = <processed git.refUpdate revision> // repo.RepoSha = <sha256 sum of git refs> repo.State = models.RepoStateActive repo.ErrorMsg = "" repo.RetryCount = 0 repo.RetryAfter = 0 if err := db.UpsertRepo(ctx, r.db, repo); err != nil { return false, fmt.Errorf("updating repo state to active %w", err) } return true, nil}
type knotStatusError struct { StatusCode int}
func (ke *knotStatusError) Error() string { return fmt.Sprintf("request failed with status code (HTTP %d)", ke.StatusCode)}
func isRateLimitError(err error) bool { var knotErr *knotStatusError if errors.As(err, &knotErr) { return knotErr.StatusCode == http.StatusTooManyRequests } return false}
func (r *Resyncer) checkKnot(ctx context.Context, repo *models.Repo) (models.ObjectFormat, error) { repoUrl, err := makeRepoRemoteUrl(repo.KnotDomain, repo.RepoIdentifier(), r.cfg.KnotUseSSL) if err != nil { return "", err }
repoUrl += "/info/refs?service=git-upload-pack"
r.logger.Debug("checking knot reachability", "url", repoUrl)
req, err := http.NewRequestWithContext(ctx, "GET", repoUrl, nil) if err != nil { return "", err } req.Header.Set("User-Agent", "git/2.x") req.Header.Set("Accept", "*/*")
resp, err := r.httpClient.Do(req) if err != nil { var uerr *url.Error if errors.As(err, &uerr) { return "", fmt.Errorf("request failed: %w", uerr.Unwrap()) } return "", fmt.Errorf("request failed: %w", err) } defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { return "", &knotStatusError{resp.StatusCode} }
// check if target is git server ct := resp.Header.Get("Content-Type") if !strings.Contains(ct, "application/x-git-upload-pack-advertisement") { return "", fmt.Errorf("unexpected content-type: %s", ct) }
advertisement, err := io.ReadAll(io.LimitReader(resp.Body, 64*1024)) if err != nil { return "", fmt.Errorf("reading upload-pack advertisement: %w", err) } if bytes.Contains(advertisement, []byte("object-format=sha256")) { return models.ObjectFormatSHA256, nil }
return models.ObjectFormatSHA1, nil}
func (r *Resyncer) suspendUnsupported(ctx context.Context, repo *models.Repo) (bool, error) { if err := r.gitm.Delete(repo); err != nil { r.logger.Warn("failed to remove local clone of suspended repo", "did", repo.RepoDid, "err", err) }
repo.State = models.RepoStateSuspended repo.ErrorMsg = "unsupported sha256 object format" repo.RetryCount = 0 repo.RetryAfter = 0 if err := db.UpsertRepo(ctx, r.db, repo); err != nil { return false, fmt.Errorf("suspending sha256 repo: %w", err) }
r.logger.Info("suspended sha256 repo, reads forwarded to knot", "did", repo.RepoDid, "knot", repo.KnotDomain) return true, nil}
func (r *Resyncer) handleResyncFailure(ctx context.Context, repoDid syntax.DID, err error) error { r.logger.Debug("handleResyncFailure", "at_uri", repoDid, "err", err) var state models.RepoState var errMsg string if err == nil { state = models.RepoStateDesynchronized errMsg = "" } else { state = models.RepoStateError errMsg = err.Error() }
repo, err := db.GetRepoByRepoDid(ctx, r.db, repoDid) if err != nil { return fmt.Errorf("failed to get repo: %w", err) } if repo == nil { return fmt.Errorf("failed to get repo. repo '%s' doesn't exist in db", repoDid) }
// start a 1 min & go up to 1 hr between retries var retryCount = repo.RetryCount + 1 var retryAfter = time.Now().Add(backoff(retryCount, 60) * 60).Unix()
// remove null bytes errMsg = strings.ReplaceAll(errMsg, "\x00", "")
repo.State = state repo.ErrorMsg = errMsg repo.RetryCount = retryCount repo.RetryAfter = retryAfter if err := db.UpsertRepo(ctx, r.db, repo); err != nil { return fmt.Errorf("failed to update repo state: %w", err) } return nil}
func backoff(retries int, max int) time.Duration { dur := min(1<<retries, max) jitter := time.Millisecond * time.Duration(rand.Intn(1000)) return time.Second*time.Duration(dur) + jitter}
func (r *Resyncer) requestIndex(ctx context.Context, repoDid syntax.DID, branches []branch) error { r.logger.Info("requesting index", "repo", repoDid, "branches", branches) body, err := json.Marshal(map[string]any{ "repo": repoDid.String(), "branches": branches, }) if err != nil { return fmt.Errorf("marshaling index request: %w", err) }
endpoint := r.cfg.Search.ZoektUrl + "/admin/enqueueIndex" req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) if err != nil { return err } req.Header.Set("Content-Type", "application/json")
resp, err := r.httpClient.Do(req) if err != nil { return fmt.Errorf("requesting zoekt index: %w", err) } defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 { return fmt.Errorf("non-ok status: %d", resp.StatusCode) } return nil}