diff --git a/spindle/notify_event.go b/spindle/notify_event.go index abeb6cab9..2887f0bb5 100644 --- a/spindle/notify_event.go +++ b/spindle/notify_event.go @@ -30,21 +30,18 @@ import ( const ( notifyWorkBudget = 5 * time.Minute - // a hostile knot can sit on its slots for the whole budget, so a knot only keeps the pool to itself - // while nobody else is waiting. a turned away knot retries with a backoff of up to a minute, so it - // counts as waiting until it gets in, or for twice that if it never comes back + // a turned away knot counts as waiting for two minutes, past its retry backoff defaultNotifySlots = 64 notifyMinShare = 2 notifyWaitWindow = 2 * time.Minute - // repos nobody serves yet cost a knot read and a pds read before anything else, so they get a - // small pool of their own and can't crowd out the repos that are served + // a repo nobody serves costs two reads before anything else, so unknown repos get their own pool notifyMaxUnknown = 2 notifyMaxUnknownPerKnot = 1 unknownRepoMemory = time.Minute unknownRepoMemoryMax = 4096 - // the knot's outbox gives up on an event after 3 days, so a push is never replayed past that + // the knot's outbox gives up after 3 days, so remember a push a bit longer than that pushMemory = 3*24*time.Hour + time.Hour changedFileOverheadBytes = 48 @@ -52,7 +49,7 @@ const ( changedFilesMaxCount = 8192 ) -// carries the status the knot should see, xrpc picks it up without importing this package +// xrpc finds the status through HTTPStatus, so it doesn't import this package type notifyStatus struct { status int err error @@ -62,7 +59,7 @@ func (e notifyStatus) Error() string { return e.err.Error() } func (e notifyStatus) Unwrap() error { return e.err } func (e notifyStatus) HTTPStatus() int { return e.status } -// the repos table and the knot's did doc spell a host differently, and every cap is per host +// the repos table and did docs spell a host differently, so every cap keys on this form func canonicalKnotHost(host string) string { name, port, hasPort := strings.Cut(strings.ToLower(host), ":") name = strings.TrimSuffix(name, ".") @@ -72,21 +69,17 @@ func canonicalKnotHost(host string) string { return name } -// what a request for a slot is measured against type slotLimits struct { pool int - // never taken by one knot alone, so a newcomer finds it on an idle pool + // one knot alone can't take this, so a newcomer always finds a free slot reserve int - // slots only the priority knots can take dedicated int priority map[string]bool } -// pool/8 with a floor of 2 func slotReserve(pool int) int { return max(2, pool/8) } -// the pool is shared by the knots that have work, so it only holds knots with work running or -// waiting, and it can't grow with the knots seen +// it only holds knots with work, so it can't grow with every knot seen type notifySlots struct { mu sync.Mutex held map[string]int @@ -105,9 +98,7 @@ func (n *notifySlots) clock() time.Time { return time.Now() } -// a priority knot takes a dedicated slot first, and every knot then goes by the shared rules. a knot may -// take all but the reserve of the pool while no other knot is waiting. once others are, each gets an -// equal share of it, and a knot at its share is turned away and counted as waiting itself +// a turned away knot counts as waiting, so it splits the pool with the rest func (n *notifySlots) acquire(host string, lim slotLimits) bool { host = canonicalKnotHost(host) n.mu.Lock() @@ -144,7 +135,6 @@ func (n *notifySlots) acquire(host string, lim slotLimits) bool { return true } -// a shared slot goes back first, it's the one other knots wait for func (n *notifySlots) release(host string) { host = canonicalKnotHost(host) n.mu.Lock() @@ -176,7 +166,6 @@ func (n *notifySlots) othersWaiting(host string) int { return count } -// once a window is enough, the entries are only read against the window func (n *notifySlots) forgetStale(now time.Time) { if now.Sub(n.swept) < notifyWaitWindow { return @@ -189,7 +178,6 @@ func (n *notifySlots) forgetStale(now time.Time) { } } -// the hosts of the configured priority knots, which may be named by host or by did:web func priorityKnotHosts(knots []string) map[string]bool { hosts := map[string]bool{} for _, knot := range knots { @@ -204,7 +192,6 @@ func priorityKnotHosts(knots []string) map[string]bool { return hosts } -// the pool for repos the db doesn't know, plus a short memory of the ones that turned out not served here type notifyAdmission struct { mu sync.Mutex held map[string]int @@ -267,8 +254,14 @@ func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid sy key := fmt.Sprintf("%s %s %s %s", repoDid, push.Ref, push.OldSha, push.NewSha) // the work outlives a knot hanging up, so its retry joins it instead of redoing it res := s.pushGroup.DoChan(key, func() (any, error) { + s.notifyWork.Add(1) + defer s.notifyWork.Done() work, cancel := context.WithTimeout(context.WithoutCancel(ctx), notifyWorkBudget) defer cancel() + // shutdown cancels the work, or git keeps writing into repos after the db closes + if s.rootCtx != nil { + defer context.AfterFunc(s.rootCtx, cancel)() + } return s.runPush(work, k, repoDid, push) }) select { @@ -280,7 +273,6 @@ func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid sy } } -// the budget for events that fetch from a knot, shared fairly between the knots that have work func (s *Spindle) acquireNotifySlot(host string) (release func(), ok bool) { if !s.notifySlots.acquire(host, s.notifyLimits()) { return nil, false @@ -311,7 +303,6 @@ func (s *Spindle) runPush(ctx context.Context, k knotclient.Knot, repoDid syntax log.FromContext(ctx).Error("notify event failed", "repo", repoDid, "err", err) } }() - // whichever fetch the knot refused, a retry gets the same answer defer func() { err = refusedAsStatus(err) }() repo, err := s.repoForPush(ctx, k, repoDid) @@ -327,7 +318,7 @@ func (s *Spindle) runPush(ctx context.Context, k knotclient.Knot, repoDid syntax return s.handlePush(ctx, k, repo, push) } -// a repo in the db already passed the manifest check in ensureRepo, so only the others read the knot +// the db only has repos that passed ensureRepo, so only unknown ones read the knot func (s *Spindle) repoForPush(ctx context.Context, k knotclient.Knot, repoDid syntax.DID) (*db.Repo, error) { repo, err := s.db.GetRepoByDid(repoDid) if err == nil { @@ -355,7 +346,7 @@ func (s *Spindle) repoForPush(ctx context.Context, k knotclient.Knot, repoDid sy return repo, nil } -// the knot drops what a 4xx refuses, so a refusal a retry would only repeat gets one +// the knot drops an event on any 4xx, so a refusal a retry can't fix gets one func refusedAsStatus(err error) error { if errors.Is(err, errRevRefused) || errors.Is(err, errTooManyWorkflowFiles) { return notifyStatus{http.StatusUnprocessableEntity, err} @@ -373,21 +364,21 @@ func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Re oldSha, oldOk := knotfeed.ParseObjectID(push.OldSha) newSha, newOk := knotfeed.ParseObjectID(push.NewSha) if !oldOk || !newOk { - // a retry sends the same shas + // a retry sends the same shas, so the 422 is final return "", notifyStatus{http.StatusUnprocessableEntity, fmt.Errorf("push shas %q and %q are not object ids", push.OldSha, push.NewSha)} } if newSha.IsZero() { l.Info("ref deleted, nothing to trigger") return "", nil } - // the knot says who pushed, and a banned user's push runs nothing even on a repo that isn't banned + // the knot says who pushed, so a banned user's push runs nothing even on an unbanned repo if ban, err := s.db.GetBan(syntax.DID(*push.Actor)); err != nil { return "", fmt.Errorf("checking bans: %w", err) } else if ban != nil { l.Warn("push by a banned user, running nothing", "actor", *push.Actor) return "", nil } - // a banned repo or a paused owner gets nothing, and it's cheaper to say so before the fetch + // checked before the fetch, so a banned repo costs no git work if gate, err := s.db.Gate(&s.cfg.Server, repo.RepoDid); err != nil { return "", fmt.Errorf("checking admission: %w", err) } else if gate.Banned || gate.Paused { @@ -438,7 +429,7 @@ func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Re return pipelineURI(k.Host, pipelineId), nil } -// the repo lock covers only these db reads and the push webhook, the fetch after them can hang on a knot +// the repo lock only covers these db reads and the webhook, the fetch can hang on a knot func (s *Spindle) admitPush(l *slog.Logger, k knotclient.Knot, repo *db.Repo, push *org_tangled.EventPush, oldSha, newSha knotfeed.ObjectID, fire func()) (uri syntax.ATURI, done bool, err error) { defer s.lockRepo(repo.RepoDid).Unlock() @@ -448,20 +439,19 @@ func (s *Spindle) admitPush(l *slog.Logger, k knotclient.Knot, repo *db.Repo, pu l.Info("repo went away before its push ran, dropping the push") return "", true, nil } - // the push is announced the first time it's seen, whatever happens to it after, and never again + // the webhook fires on the first recording, even if the push fails later if first, err := s.db.RecordPush(repo.RepoDid, push.Ref, oldSha.String(), newSha.String(), time.Now()); err != nil { return "", true, fmt.Errorf("recording the push: %w", err) } else if first { fire() } - // a replay of a push that made a pipeline runs nothing if id, found, err := s.db.PushPipeline(repo.RepoDid.String(), push.Ref, oldSha.String(), newSha.String()); err != nil { return "", true, fmt.Errorf("looking up the push: %w", err) } else if found { l.Info("push already has a pipeline, skipping the replay", "pipeline", id) return pipelineURI(k.Host, id), true, nil } - // skip ci only skips the pipeline, the webhook already went out + // skip ci only skips the pipeline, the webhook already fired above if kgit.HasSkipCIPushOption(push.PushOptions) { l.Info("push requested ci skip, skipping the event") return "", true, nil @@ -490,8 +480,7 @@ func isMaterializedRef(refname string) bool { return strings.HasPrefix(refname, "refs/heads/") || strings.HasPrefix(refname, "refs/tags/") } -// known is false when the changed files can't be told, and the nil list then makes the compiler run -// the workflows a paths filter would have gated, because missing CI is worse than extra CI +// unknown changes run the workflows a paths filter would gate, since missing ci is worse func changedFilesUnderBudget(ctx context.Context, l *slog.Logger, repoPath string, oldSha, newSha knotfeed.ObjectID) (files []string, known bool) { if oldSha.IsZero() { return nil, false @@ -509,7 +498,7 @@ func changedFilesUnderBudget(ctx context.Context, l *slog.Logger, repoPath strin return admitted, true } -// the sync is shallow, so a missing oldSha gets fetched here, and go-git can't do sha256 +// the sync is shallow, so oldSha may be missing, and go-git can't read sha256 repos func changedFiles(ctx context.Context, repoPath string, oldSha, newSha knotfeed.ObjectID) ([]string, error) { diff := func() ([]byte, error) { return gitutil.Output(ctx, "-C", repoPath, "diff", "--name-only", "--no-renames", "-z", oldSha.String(), newSha.String()) @@ -526,7 +515,6 @@ func changedFiles(ctx context.Context, repoPath string, oldSha, newSha knotfeed. return strings.FieldsFunc(string(out), func(r rune) bool { return r == 0 }), nil } -// false when the listing is over budget, an empty list is a push that changed nothing func admitChangedFiles(paths []string) (admitted []string, ok bool) { spent := 0 admitted = make([]string, 0, min(len(paths), changedFilesMaxCount)) diff --git a/spindle/notify_shutdown_test.go b/spindle/notify_shutdown_test.go new file mode 100644 index 000000000..3a0c0e077 --- /dev/null +++ b/spindle/notify_shutdown_test.go @@ -0,0 +1,30 @@ +package spindle + +import ( + "testing" + "time" +) + +func TestShutdownEndsAPushFetchInFlight(t *testing.T) { + e := newPushEnv(t) + push := e.firstPush(t) + release := e.w.holdGit() + defer release(false) + + answered := make(chan error, 1) + go func() { + _, err := e.notify(e.n.s, push) + answered <- err + }() + e.waitClone(t) + + e.n.s.rootCancel() + select { + case err := <-answered: + if err == nil { + t.Fatal("the push was handled although its fetch was still held") + } + case <-time.After(5 * time.Second): + t.Fatal("the push's fetch kept running after the spindle was shut down") + } +} diff --git a/spindle/server.go b/spindle/server.go index 7f1881a3a..9d2723c14 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -124,6 +124,7 @@ type Spindle struct { pushGroup singleflight.Group notifySlots notifySlots + notifyWork sync.WaitGroup notifyPriorityOnce sync.Once notifyPriorityHosts map[string]bool notifyAdmission notifyAdmission @@ -536,6 +537,17 @@ func (s *Spindle) Start(ctx context.Context) error { shutdownCtx, cancelShutdown := context.WithTimeout(context.Background(), serverShutdownTimeout) shutdownErr := server.Shutdown(shutdownCtx) cancelShutdown() + // notify work can outlive the handler, so wait for it before the run ends + notifyDone := make(chan struct{}) + go func() { + s.notifyWork.Wait() + close(notifyDone) + }() + select { + case <-notifyDone: + case <-time.After(5 * time.Second): + s.l.Warn("timed out waiting for notify work to stop") + } if listenErr == nil { listenErr = <-serverErr } @@ -797,8 +809,7 @@ func Run(ctx context.Context) error { return s.Start(ctx) } -// a client that dribbles its headers or body can't hold a connection forever. no write timeout -// because log streams write for hours, and gorilla clears the read deadline on a websocket upgrade +// no write timeout since log streams run for hours, and a websocket upgrade clears the read deadline func newHTTPServer(addr string, handler http.Handler) *http.Server { return &http.Server{ Addr: addr, @@ -937,7 +948,7 @@ func (s *Spindle) resolveSourceRepoInfo(ctx context.Context, repoDid syntax.DID) func (s *Spindle) runPipeline(ctx context.Context, pipelineId models.PipelineId, repoDid syntax.DID, trigger tangled.Pipeline_TriggerMetadata, changedFiles []string, repoCloneUri, repoPath, rev string, only []string, sourceRepo *tangled.Pipeline_TriggerRepo) (models.PipelineId, error) { l := log.FromContext(ctx) - // every push, pull, manual and scheduled pipeline comes through here + // all triggers go through this gate, not just pushes if gate, err := s.db.Gate(&s.cfg.Server, repoDid); err != nil { return "", fmt.Errorf("checking admission: %w", err) } else if gate.Paused { @@ -979,12 +990,12 @@ func (s *Spindle) runPipeline(ctx context.Context, pipelineId models.PipelineId, return "", fmt.Errorf("creating pipeline: %w", err) } err = s.processPipeline(ctx, repoDid, tpl, record, sourceRepo) - // the pause or a wipe landed while the checkout ran, so nothing was enqueued + // a pause or wipe can land during the checkout, so either error means nothing to enqueue if errors.Is(err, db.ErrOwnerPaused) || errors.Is(err, db.ErrNotServed) { l.Info("nothing enqueued, the owner is paused or the repo is gone", "repo", repoDid) return "", nil } - // the id only means something once the pipeline row exists + // the id is only real once the pipeline row exists if err != nil { return "", err } @@ -1021,7 +1032,7 @@ func (s *Spindle) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, } else if ban != nil { return "", fmt.Errorf("repo %s is banned", repoDid) } - // before any checkout, and the caller is told why instead of being told nothing matched + // checked before the checkout, so a paused owner gets an error and not an empty match if gate, err := s.db.Gate(&s.cfg.Server, repoDid); err != nil { return "", fmt.Errorf("checking admission: %w", err) } else if gate.Paused { @@ -1153,7 +1164,7 @@ func (s *Spindle) loadWorkflowDefinitions(ctx context.Context, repoDid syntax.DI } func (s *Spindle) DescribeWorkflowDefinition(ctx context.Context, repoDid syntax.DID, sha string, sourceRepo syntax.DID) (*tangled.CiDescribeWorkflowDefinition_Output, error) { - // anyone can ask, so a source repo has to be one this spindle already serves, not any did + // anyone can ask, so the source repo must be one we already serve if sourceRepo != "" && sourceRepo != repoDid { if _, err := s.db.GetRepoByDid(sourceRepo); errors.Is(err, sql.ErrNoRows) { return nil, xrpc.ErrSourceRepoNotServed @@ -1231,11 +1242,11 @@ var ( // a 4xx means the knot won't serve that commit, a retry gets the same answer unless it was busy func refusedByKnot(err error) bool { msg := err.Error() - // upload-pack says no to an unknown sha inside a 200, so there is no status to read + // upload-pack reports an unknown sha as a 200, so there's no status code to check if strings.Contains(msg, "not our ref") || strings.Contains(msg, "unadvertised object") { return true } - // a plain 404 on info/refs comes out as this, without the status + // a plain 404 on info/refs only shows up in the message text if strings.Contains(msg, "fatal: repository '") && strings.Contains(msg, "' not found") { return true } @@ -1352,7 +1363,7 @@ func (s *Spindle) StartJobWorkers(ctx context.Context) { s.l.Error("failed to dequeue job", "error", err) } if job == nil { - // a failed dequeue can leave jobs queued that no new enqueue will wake us for + // a failed dequeue can leave jobs queued with nothing to wake us, so retry on a timer var retry <-chan time.Time if err != nil { retry = time.After(dequeueRetryDelay) @@ -1547,7 +1558,7 @@ func (s *Spindle) idTokenMinter(job *db.JobRow, trustedSource bool) func(time.Du func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, record *models.PipelineRecord, sourceRepo *tangled.Pipeline_TriggerRepo) error { traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx) - // only the insert is under the repo lock, so a wipe can't land half way through it + // only the insert holds the repo lock, so a wipe can't land in the middle of it err := func() error { defer s.lockRepo(repoDid).Unlock() return s.db.EnqueuePipeline(s.rootCtx, repoDid.String(), record, sourceRepo, tpl, traceparent, tracestate, s.n, db.Admitted(&s.cfg.Server))