From f9cce9b25d2b9a367d50e757e097de326d9baf41 Mon Sep 17 00:00:00 2001 From: dawn Date: Thu, 22 Oct 2026 04:03:04 +0300 Subject: [PATCH] jetstream,spindle: drop duplicated code Signed-off-by: dawn --- jetstream/jetstream.go | 2 +- spindle/db/db.go | 35 ++-- spindle/db/pipelines.go | 18 ++- spindle/db/quota.go | 264 +++++++++++-------------------- spindle/db/repo_retries.go | 19 +-- spindle/db/wipes.go | 13 +- spindle/hydrant.go | 19 +-- spindle/knotclient/knotclient.go | 2 +- spindle/notify_event.go | 41 +++-- spindle/pulls.go | 28 ++-- spindle/schedule.go | 25 ++- spindle/wipe.go | 42 ++--- spindle/workflow_defs.go | 37 ++--- 13 files changed, 208 insertions(+), 337 deletions(-) diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go index 04de17b2d..95fc60fb4 100644 --- a/jetstream/jetstream.go +++ b/jetstream/jetstream.go @@ -43,7 +43,7 @@ type JetstreamClient struct { lastSeenUs atomic.Int64 marks watermark - holding atomic.Int64 + holding atomic.Int64 // the first event of each connection clears it, so the gap check runs once anchorUs atomic.Int64 diff --git a/spindle/db/db.go b/spindle/db/db.go index 453122c55..0e48977b2 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -55,7 +55,7 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { if err != nil { return nil, err } - // viewers that cancel mid-query would churn fds, so keep the idle conns around + // cancelled viewer queries would churn fds unless idle conns stick around db.SetMaxIdleConns(maxIdleConns) conn, err := db.Conn(ctx) @@ -1193,26 +1193,15 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } if err := orm.RunMigration(conn, logger, "repo-retries-epoch", func(tx *sql.Tx) error { - present, err := tableHasColumn(tx, "repo_retries", "epoch") - if err != nil || present { - return err - } - _, err = tx.Exec(`alter table repo_retries add column epoch integer not null default 0`) - return err + return addColumnIfMissing(tx, "repo_retries", "epoch", "integer not null default 0") }); err != nil { return err } if err := orm.RunMigration(conn, logger, "record-retries-epoch", func(tx *sql.Tx) error { - present, err := tableHasColumn(tx, "record_retries", "epoch") - if err != nil { + if err := addColumnIfMissing(tx, "record_retries", "epoch", "integer not null default 0"); err != nil { return err } - if !present { - if _, err := tx.Exec(`alter table record_retries add column epoch integer not null default 0`); err != nil { - return err - } - } - _, err = tx.Exec(` + _, err := tx.Exec(` create index if not exists idx_record_retries_due on record_retries(due); create index if not exists idx_repo_retries_due on repo_retries(due); `) @@ -1221,12 +1210,7 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } if err := orm.RunMigration(conn, logger, "pull-ticket-version", func(tx *sql.Tx) error { - present, err := tableHasColumn(tx, "pull_tickets", "version") - if err != nil || present { - return err - } - _, err = tx.Exec(`alter table pull_tickets add column version text not null default ''`) - return err + return addColumnIfMissing(tx, "pull_tickets", "version", "text not null default ''") }); err != nil { return err } @@ -1312,6 +1296,15 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return nil } +func addColumnIfMissing(tx *sql.Tx, table, column, definition string) error { + present, err := tableHasColumn(tx, table, column) + if err != nil || present { + return err + } + _, err = tx.Exec("alter table " + table + " add column " + column + " " + definition) + return err +} + func tableHasColumn(tx *sql.Tx, table, column string) (bool, error) { var count int if err := tx.QueryRow( diff --git a/spindle/db/pipelines.go b/spindle/db/pipelines.go index 900faaf95..9140922dc 100644 --- a/spindle/db/pipelines.go +++ b/spindle/db/pipelines.go @@ -33,28 +33,30 @@ func (d *DB) QueryPipelines(ctx context.Context, repoDID string, commits []strin } whereClause := " where " + strings.Join(conditions, " and ") - baseWhere, baseArgs := whereClause, args - - // the cursor carries the total, so later pages skip the count + // later pages skip the count since the cursor carries the total var total int64 counted := false + cursorID, hasCursor := int64(0), false if cursor != "" { idText, totalText, _ := strings.Cut(cursor, ":") if value, err := strconv.ParseInt(idText, 10, 64); err == nil { - filter := orm.FilterLt("id", value) - conditions = append(conditions, filter.Condition()) - args = append(args, filter.Arg()...) - whereClause = " where " + strings.Join(conditions, " and ") + cursorID, hasCursor = value, true if carried, err := strconv.ParseInt(totalText, 10, 64); err == nil { total, counted = carried, true } } } if !counted { - if err := d.QueryRowContext(ctx, `select count(*) from pipelines`+baseWhere, baseArgs...).Scan(&total); err != nil { + if err := d.QueryRowContext(ctx, `select count(*) from pipelines`+whereClause, args...).Scan(&total); err != nil { return nil, "", 0, err } } + if hasCursor { + filter := orm.FilterLt("id", cursorID) + conditions = append(conditions, filter.Condition()) + args = append(args, filter.Arg()...) + whereClause = " where " + strings.Join(conditions, " and ") + } args = append(args, limit) rows, err := d.QueryContext(ctx, diff --git a/spindle/db/quota.go b/spindle/db/quota.go index 42d02b8a7..68a31fa03 100644 --- a/spindle/db/quota.go +++ b/spindle/db/quota.go @@ -126,7 +126,7 @@ func (d *QuotaStore) Reserve(ctx context.Context, req quota.ReserveRequest) (quo return res, nil } -// same decision as Reserve without the write lock, so a polling waiter stays off it +// same check as Reserve minus the write lock, which keeps pollers off it func (d *QuotaStore) CheckDenied(ctx context.Context, req quota.ReserveRequest) (quota.Reservation, bool, error) { if err := quota.Validate(req); err != nil { return quota.Reservation{}, false, err @@ -145,7 +145,6 @@ func (d *QuotaStore) CheckDenied(ctx context.Context, req quota.ReserveRequest) } func (d *QuotaStore) reserveOn(ctx context.Context, conn *sql.Conn, req quota.ReserveRequest, dryRun bool) (quota.Reservation, error) { - var res quota.Reservation if !dryRun { _, err := conn.ExecContext(ctx, ` INSERT INTO quota_repo_owners (repo_did, owner_did) @@ -153,96 +152,36 @@ func (d *QuotaStore) reserveOn(ctx context.Context, conn *sql.Conn, req quota.Re ON CONFLICT(repo_did) DO UPDATE SET owner_did = excluded.owner_did `, req.Identity.RepoDID, req.Identity.OwnerDID) if err != nil { - return res, err + return quota.Reservation{}, err } } - rows, err := conn.QueryContext(ctx, ` + live, err := queryAmountsByID(ctx, conn, ` SELECT id, resource, amount FROM quota_reservations WHERE repo_did = ? AND kind = ? AND key = ? AND phase IN ('reserved', 'publishing', 'active') `, req.Identity.RepoDID, string(req.Kind), req.Key) if err != nil { - return res, err + return quota.Reservation{}, err } - defer rows.Close() - - resMapByID := make(map[string]map[string]int64) - for rows.Next() { - var id, resourceStr string - var amount int64 - if err := rows.Scan(&id, &resourceStr, &amount); err != nil { - return res, err - } - if _, ok := resMapByID[id]; !ok { - resMapByID[id] = make(map[string]int64) - } - resMapByID[id][resourceStr] = amount - } - for id, resources := range resMapByID { - if len(resources) == len(req.Resources) { - match := true - for r, amt := range req.Resources { - if resources[r] != amt { - match = false - break - } - } - if match { - res = quota.Reservation{ - ID: id, - Allowed: true, - Temporary: false, - Reason: quota.ReasonUnlimited, - } - return res, nil - } + for id, amounts := range live { + if sameAmounts(amounts, req.Resources) { + return quota.Reservation{ID: id, Allowed: true, Reason: quota.ReasonUnlimited}, nil } } - allocRows, err := conn.QueryContext(ctx, ` - SELECT resource, amount FROM quota_allocations + allocated, err := queryAmountsByID(ctx, conn, ` + SELECT '', resource, amount FROM quota_allocations WHERE repo_did = ? AND kind = ? AND key = ? `, req.Identity.RepoDID, string(req.Kind), req.Key) if err != nil { - return res, err - } - defer allocRows.Close() - - allocResources := make(map[string]int64) - for allocRows.Next() { - var resourceStr string - var amount int64 - if err := allocRows.Scan(&resourceStr, &amount); err != nil { - return res, err - } - allocResources[resourceStr] = amount + return quota.Reservation{}, err } - if len(allocResources) == len(req.Resources) { - match := true - for r, amt := range req.Resources { - if allocResources[r] != amt { - match = false - break - } - } - if match { - res = quota.Reservation{ - ID: "", - Allowed: true, - Temporary: false, - Reason: quota.ReasonUnlimited, - } - return res, nil - } + if sameAmounts(allocated[""], req.Resources) { + return quota.Reservation{Allowed: true, Reason: quota.ReasonUnlimited}, nil } - deny := func(reason, resource string, temporary bool) { - res = quota.Reservation{ - Allowed: false, - Temporary: temporary, - Reason: reason, - Resource: resource, - } + deny := func(reason, resource string, temporary bool) quota.Reservation { + return quota.Reservation{Allowed: false, Temporary: temporary, Reason: reason, Resource: resource} } var sortedResources []string @@ -254,29 +193,10 @@ func (d *QuotaStore) reserveOn(ctx context.Context, conn *sql.Conn, req quota.Re for _, resource := range sortedResources { amount := req.Resources[resource] - var repoLimit *int64 - var maxAmount *int64 - err = conn.QueryRowContext(ctx, ` - SELECT max_amount FROM quota_limits - WHERE did = ? AND resource = ? - `, req.Identity.RepoDID, resource).Scan(&maxAmount) - if err == nil { - if maxAmount != nil { - val := *maxAmount - repoLimit = &val - } - } else if errors.Is(err, sql.ErrNoRows) { - if d.defaults != nil { - if resMap, ok := d.defaults[quota.ScopeRepo]; ok { - if val, ok := resMap[resource]; ok && val > 0 { - repoLimit = &val - } - } - } - } else { - return res, err + repoLimit, err := d.limitFor(ctx, conn, quota.ScopeRepo, req.Identity.RepoDID, resource) + if err != nil { + return quota.Reservation{}, err } - currentRepoUsage, _, err := sumQuotaUsage(ctx, conn, string(req.Kind), req.Key, amount, ` SELECT kind, key, amount FROM ( SELECT kind, key, amount FROM quota_allocations WHERE repo_did = ? AND resource = ? @@ -285,40 +205,19 @@ func (d *QuotaStore) reserveOn(ctx context.Context, conn *sql.Conn, req quota.Re ) `, req.Identity.RepoDID, resource, req.Identity.RepoDID, resource) if err != nil { - return res, err + return quota.Reservation{}, err } if amount > math.MaxInt64-currentRepoUsage { - deny(quota.ReasonRepoLimit, resource, true) - return res, nil + return deny(quota.ReasonRepoLimit, resource, true), nil } if repoLimit != nil && *repoLimit >= 0 && amount > *repoLimit-currentRepoUsage { - deny(quota.ReasonRepoLimit, resource, amount <= *repoLimit) - return res, nil - } - - var userLimit *int64 - var maxUserAmount *int64 - err = conn.QueryRowContext(ctx, ` - SELECT max_amount FROM quota_limits - WHERE did = ? AND resource = ? - `, req.Identity.OwnerDID, resource).Scan(&maxUserAmount) - if err == nil { - if maxUserAmount != nil { - val := *maxUserAmount - userLimit = &val - } - } else if errors.Is(err, sql.ErrNoRows) { - if d.defaults != nil { - if resMap, ok := d.defaults[quota.ScopeUser]; ok { - if val, ok := resMap[resource]; ok && val > 0 { - userLimit = &val - } - } - } - } else { - return res, err + return deny(quota.ReasonRepoLimit, resource, amount <= *repoLimit), nil } + userLimit, err := d.limitFor(ctx, conn, quota.ScopeUser, req.Identity.OwnerDID, resource) + if err != nil { + return quota.Reservation{}, err + } currentUserUsage, existsOwner, err := sumQuotaUsage(ctx, conn, string(req.Kind), req.Key, amount, ` SELECT kind, key, amount FROM ( SELECT a.kind, a.key, a.amount @@ -332,7 +231,7 @@ func (d *QuotaStore) reserveOn(ctx context.Context, conn *sql.Conn, req quota.Re ) `, req.Identity.OwnerDID, resource, req.Identity.OwnerDID, resource) if err != nil { - return res, err + return quota.Reservation{}, err } var additional int64 @@ -340,25 +239,22 @@ func (d *QuotaStore) reserveOn(ctx context.Context, conn *sql.Conn, req quota.Re additional = amount } if additional > math.MaxInt64-currentUserUsage { - deny(quota.ReasonUserLimit, resource, true) - return res, nil + return deny(quota.ReasonUserLimit, resource, true), nil } - if userLimit != nil && *userLimit >= 0 && additional > 0 && additional > *userLimit-currentUserUsage { - deny(quota.ReasonUserLimit, resource, additional <= *userLimit) - return res, nil + return deny(quota.ReasonUserLimit, resource, additional <= *userLimit), nil } } if dryRun { - return res, nil + return quota.Reservation{Allowed: true}, nil } resID := req.ID if resID == "" { var b [16]byte - if _, err = rand.Read(b[:]); err != nil { - return res, err + if _, err := rand.Read(b[:]); err != nil { + return quota.Reservation{}, err } resID = hex.EncodeToString(b[:]) } @@ -369,57 +265,87 @@ func (d *QuotaStore) reserveOn(ctx context.Context, conn *sql.Conn, req quota.Re } for _, resource := range sortedResources { - amount := req.Resources[resource] _, err = conn.ExecContext(ctx, ` INSERT INTO quota_reservations (id, resource, kind, key, amount, repo_did, owner_did, phase, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - `, resID, resource, string(req.Kind), req.Key, amount, req.Identity.RepoDID, req.Identity.OwnerDID, phase, time.Now().Unix()) + `, resID, resource, string(req.Kind), req.Key, req.Resources[resource], req.Identity.RepoDID, req.Identity.OwnerDID, phase, time.Now().Unix()) if err != nil { - return res, err - } - } - - hasFiniteLimit := func(scope quota.Scope, did string, resource string) (bool, error) { - var override sql.NullInt64 - err := conn.QueryRowContext(ctx, ` - SELECT max_amount FROM quota_limits - WHERE did = ? AND resource = ? - `, did, resource).Scan(&override) - switch { - case err == nil: - return override.Valid, nil - case !errors.Is(err, sql.ErrNoRows): - return false, err - case d.defaults == nil || d.defaults[scope] == nil: - return false, nil - default: - return d.defaults[scope][resource] > 0, nil + return quota.Reservation{}, err } } reason := quota.ReasonUnlimited for _, resource := range sortedResources { - repoFinite, err := hasFiniteLimit(quota.ScopeRepo, req.Identity.RepoDID, resource) + repoLimit, err := d.limitFor(ctx, conn, quota.ScopeRepo, req.Identity.RepoDID, resource) if err != nil { - return res, err + return quota.Reservation{}, err } - userFinite, err := hasFiniteLimit(quota.ScopeUser, req.Identity.OwnerDID, resource) + userLimit, err := d.limitFor(ctx, conn, quota.ScopeUser, req.Identity.OwnerDID, resource) if err != nil { - return res, err + return quota.Reservation{}, err } - if repoFinite || userFinite { + if repoLimit != nil || userLimit != nil { reason = quota.ReasonWithinLimit break } } + return quota.Reservation{ID: resID, Allowed: true, Reason: reason}, nil +} - res = quota.Reservation{ - ID: resID, - Allowed: true, - Temporary: false, - Reason: reason, +// a null override means unlimited, not the default +func (d *QuotaStore) limitFor(ctx context.Context, conn *sql.Conn, scope quota.Scope, did, resource string) (*int64, error) { + var override sql.NullInt64 + err := conn.QueryRowContext(ctx, ` + SELECT max_amount FROM quota_limits + WHERE did = ? AND resource = ? + `, did, resource).Scan(&override) + switch { + case err == nil: + if !override.Valid { + return nil, nil + } + return &override.Int64, nil + case !errors.Is(err, sql.ErrNoRows): + return nil, err } - return res, nil + if val := d.defaults[scope][resource]; val > 0 { + return &val, nil + } + return nil, nil +} + +func queryAmountsByID(ctx context.Context, conn *sql.Conn, query string, args ...any) (map[string]map[string]int64, error) { + rows, err := conn.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make(map[string]map[string]int64) + for rows.Next() { + var id, resource string + var amount int64 + if err := rows.Scan(&id, &resource, &amount); err != nil { + return nil, err + } + if out[id] == nil { + out[id] = make(map[string]int64) + } + out[id][resource] = amount + } + return out, rows.Err() +} + +func sameAmounts(have, want map[string]int64) bool { + if len(have) != len(want) { + return false + } + for r, amt := range want { + if have[r] != amt { + return false + } + } + return true } func (d *QuotaStore) BeginCommit(ctx context.Context, reservationID string) error { @@ -837,7 +763,7 @@ func classifyStatus(used, limit int64) string { return "under_limit" } -// publishing reservations go too, since nothing at boot can tell if the object landed +// drop publishing rows too, since nothing at boot can tell if the object landed func (d *QuotaStore) Recover(ctx context.Context, liveIDs []string) error { return d.withTx(ctx, func(conn *sql.Conn) error { _, err := conn.ExecContext(ctx, ` @@ -870,7 +796,7 @@ func (d *QuotaStore) Recover(ctx context.Context, liveIDs []string) error { }) } -// store gc reports store paths, not nar files, so keep the path to match later +// keep the path to match on later, since gc reports store paths, not nar files func (d *QuotaStore) RecordNixStorePath(ctx context.Context, repoDID, key, storePath string) error { _, err := d.db.ExecContext(ctx, ` UPDATE quota_allocations SET store_path = ? @@ -879,7 +805,7 @@ func (d *QuotaStore) RecordNixStorePath(ctx context.Context, repoDID, key, store return err } -// the same nar can be charged to several repos, and all of them get it back +// one nar can be charged to several repos, and each one gets it back func (d *QuotaStore) ReleaseNixObjects(ctx context.Context, inputs []string) (released, bytes int64, unmatched []string, err error) { err = d.withTx(ctx, func(conn *sql.Conn) error { released, bytes, unmatched = 0, 0, nil diff --git a/spindle/db/repo_retries.go b/spindle/db/repo_retries.go index 7a3ed9473..6c8804442 100644 --- a/spindle/db/repo_retries.go +++ b/spindle/db/repo_retries.go @@ -9,11 +9,7 @@ import ( ) func (d *DB) RepoRetryAttempts(repoDid syntax.DID) (int, error) { - var attempts int - err := d.QueryRow(`select attempts from repo_retries where repo_did = ?`, repoDid).Scan(&attempts) - if errors.Is(err, sql.ErrNoRows) { - return 0, nil - } + attempts, _, err := d.RepoRetryRow(repoDid) return attempts, err } @@ -25,7 +21,6 @@ func (d *DB) RepoRetryRow(repoDid syntax.DID) (attempts int, exists bool, err er return attempts, err == nil, err } -// a fresh mark is already due, so a job only has to check due func (d *DB) RepoRetryDue(repoDid syntax.DID, now time.Time) (marked, due bool, err error) { var at int64 err = d.QueryRow(`select due from repo_retries where repo_did = ?`, repoDid).Scan(&at) @@ -35,7 +30,6 @@ func (d *DB) RepoRetryDue(repoDid syntax.DID, now time.Time) (marked, due bool, return err == nil, at <= now.Unix(), err } -// 0 when there's no row, any mark that lands after still gets a higher epoch func (d *DB) RepoRetryEpoch(repoDid syntax.DID) (int64, error) { var epoch int64 err := d.QueryRow(`select epoch from repo_retries where repo_did = ?`, repoDid).Scan(&epoch) @@ -45,10 +39,10 @@ func (d *DB) RepoRetryEpoch(repoDid syntax.DID) (int64, error) { return epoch, err } -// a failing knot can't keep its repo due by spamming events, but real news isn't stuck behind a backoff +// a failed repo's due can only move earlier, to now plus this, and spam can't stall it const failedMarkDelay = 5 * time.Second -// the epoch always moves, so a job that started earlier can't clear this mark +// every mark bumps the epoch, and a job that read the old one can't clear it func (d *DB) MarkRepoDirty(repoDid syntax.DID, now time.Time, known bool, capacity int) (bool, error) { tx, err := d.Begin() if err != nil { @@ -56,7 +50,7 @@ func (d *DB) MarkRepoDirty(repoDid syntax.DID, now time.Time, known bool, capaci } defer tx.Rollback() if !known { - // only failed rows get evicted, since those are the dids a hostile knot makes up + // only failed rows get evicted, otherwise a knot could push out real repos with made up dids if _, err := tx.Exec( `delete from repo_retries where rowid = (select rowid from repo_retries where attempts > 0 order by epoch limit 1) @@ -84,7 +78,6 @@ func (d *DB) MarkRepoDirty(repoDid syntax.DID, now time.Time, known bool, capaci return n > 0, tx.Commit() } -// a mark since the epoch was read leaves the row alone func (d *DB) ClearRepoRetryAt(repoDid syntax.DID, epoch int64) error { _, err := d.Exec(`delete from repo_retries where repo_did = ? and epoch = ?`, repoDid, epoch) return err @@ -121,7 +114,6 @@ type RecordRetry struct { Rkey string } -// epoch is what lets a job's clear skip a mark that lands later type RecordMark struct { RecordRetry Epoch int64 @@ -139,7 +131,6 @@ func (d *DB) RecordRetryEpoch(r RecordRetry) (int64, error) { return epoch, err } -// a flooding repo only evicts its own marks, never another repo's func (d *DB) MarkRecordDirty(r RecordRetry, now time.Time, perRepo int) error { tx, err := d.Begin() if err != nil { @@ -224,7 +215,7 @@ func (d *DB) DeleteRecordRetry(r RecordRetry) error { return err } -// a repo with a failed refresh retries on its own, so it's left out here +// a repo in repo_retries already gets retried whole, skip it here func (d *DB) DueRecordMarkRepos(now time.Time) ([]syntax.DID, error) { return queryDIDs(d, `select distinct repo_did from record_retries diff --git a/spindle/db/wipes.go b/spindle/db/wipes.go index f9c130349..75165bafb 100644 --- a/spindle/db/wipes.go +++ b/spindle/db/wipes.go @@ -16,7 +16,7 @@ type Wipe struct { Due time.Time } -// one tx, so a repo is never unserved without a wipe row to finish it +// one tx, or a repo could lose its rows with no wipe row to finish it // an existing wipe starts over, the repo may have come back since func (d *DB) ClaimWipe(repoDid syntax.DID, reason string, steps []string, due time.Time) error { tx, err := d.Begin() @@ -40,7 +40,7 @@ func (d *DB) ClaimWipe(repoDid syntax.DID, reason string, steps []string, due ti return tx.Commit() } -// no wipe for the repo gives sql.ErrNoRows, not nil +// no wipe row is sql.ErrNoRows, not nil func (d *DB) GetWipe(repoDid syntax.DID) (*Wipe, error) { w := Wipe{RepoDid: repoDid} var steps string @@ -86,12 +86,11 @@ func (d *DB) FinishWipeStep(repoDid syntax.DID, step string) error { } } if len(left) == 0 { - if _, err := tx.Exec(`delete from wipes where repo_did = ?`, repoDid.String()); err != nil { - return err - } - return tx.Commit() + _, err = tx.Exec(`delete from wipes where repo_did = ?`, repoDid.String()) + } else { + _, err = tx.Exec(`update wipes set steps = ? where repo_did = ?`, strings.Join(left, ","), repoDid.String()) } - if _, err := tx.Exec(`update wipes set steps = ? where repo_did = ?`, strings.Join(left, ","), repoDid.String()); err != nil { + if err != nil { return err } return tx.Commit() diff --git a/spindle/hydrant.go b/spindle/hydrant.go index 729544b92..8e1d090ba 100644 --- a/spindle/hydrant.go +++ b/spindle/hydrant.go @@ -101,14 +101,13 @@ type backfill struct { // id of the last stream event spindle finished with, -1 before the first last atomic.Int64 - // a gap mid catch-up only sets catchUpAgain, the loop runs once more + // a gap during catch-up only sets catchUpAgain, the running loop goes once more catchingUp bool catchUpAgain bool catchUps sync.WaitGroup closed bool } -// backfill events only mark repos, the repo jobs read the knot and pds func newBackfill(s *Spindle, l *slog.Logger, dir string) *backfill { return &backfill{ s: s, @@ -167,10 +166,7 @@ func (b *backfill) Track(dids ...syntax.DID) { } b.mu.Unlock() if queued { - select { - case b.wake <- struct{}{}: - default: - } + nudge(b.wake) } } @@ -198,7 +194,7 @@ func (b *backfill) CatchUp() error { return nil } -// runs on the jetstream reader, so it returns without waiting on the catch-up +// called from the jetstream reader, which can't block on the catch-up func (b *backfill) CatchUpLater() { b.mu.Lock() defer b.mu.Unlock() @@ -206,7 +202,7 @@ func (b *backfill) CatchUpLater() { b.catchUpAgain = true return } - // close waits on catchUps, so nothing new can start once closed + // close waits on catchUps, and nothing may start once closed is set if b.closed { return } @@ -228,7 +224,6 @@ func (b *backfill) CatchUpLater() { }) } -// a served repo's own manifest, collaborator or declaration record needs no repo job func (s *Spindle) markBackfilled(ctx context.Context, evt tapc.Event) error { if evt.Type == tapc.EvtRecord && evt.Record != nil { switch evt.Record.Collection.String() { @@ -296,10 +291,7 @@ func (b *backfill) queueCatchUp(dids []syntax.DID) { b.remove[did] = struct{}{} } b.mu.Unlock() - select { - case b.wake <- struct{}{}: - default: - } + nudge(b.wake) } // Untrack waits for the next time hydrant runs anyway, since forgetting an @@ -1063,7 +1055,6 @@ func (a hydrantAPI) subscribe(ctx context.Context, cursor int64) (*websocket.Con return conn, nil } -// its records were skipped on the catch-up's word, so mark it now instead of on a retry func (b *backfill) markFailedServed(settled map[syntax.DID]outcome) { for did, o := range settled { if o != failed { diff --git a/spindle/knotclient/knotclient.go b/spindle/knotclient/knotclient.go index 6d01c001a..27bb6b983 100644 --- a/spindle/knotclient/knotclient.go +++ b/spindle/knotclient/knotclient.go @@ -33,7 +33,7 @@ const ( ) var ( - ErrTooManyRecords = errors.New("more records than spindle pages through") + ErrTooManyRecords = errors.New("more records than spindle pages through") ErrNotFound = errors.New("record not found on the knot") ErrNoKnot = errors.New("repo did document names no knot") ) diff --git a/spindle/notify_event.go b/spindle/notify_event.go index 2887f0bb5..27f679b34 100644 --- a/spindle/notify_event.go +++ b/spindle/notify_event.go @@ -29,19 +29,18 @@ import ( ) const ( - notifyWorkBudget = 5 * time.Minute - // a turned away knot counts as waiting for two minutes, past its retry backoff + notifyWorkBudget = 5 * time.Minute defaultNotifySlots = 64 notifyMinShare = 2 notifyWaitWindow = 2 * time.Minute - // a repo nobody serves costs two reads before anything else, so unknown repos get their own pool + // unknown repos cost two reads each and get a pool of their own notifyMaxUnknown = 2 notifyMaxUnknownPerKnot = 1 unknownRepoMemory = time.Minute unknownRepoMemoryMax = 4096 - // the knot's outbox gives up after 3 days, so remember a push a bit longer than that + // keep pushes an hour longer because the knot's outbox gives up after 3 days pushMemory = 3*24*time.Hour + time.Hour changedFileOverheadBytes = 48 @@ -49,7 +48,7 @@ const ( changedFilesMaxCount = 8192 ) -// xrpc finds the status through HTTPStatus, so it doesn't import this package +// xrpc gets the status through HTTPStatus since it can't import spindle type notifyStatus struct { status int err error @@ -59,7 +58,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 did docs spell a host differently, so every cap keys on this form +// every cap keys on this form because the repos table and did docs spell hosts differently func canonicalKnotHost(host string) string { name, port, hasPort := strings.Cut(strings.ToLower(host), ":") name = strings.TrimSuffix(name, ".") @@ -71,15 +70,15 @@ func canonicalKnotHost(host string) string { type slotLimits struct { pool int - // one knot alone can't take this, so a newcomer always finds a free slot - reserve int + // a new knot always finds a slot because one knot can't use up the reserve + reserve int dedicated int priority map[string]bool } func slotReserve(pool int) int { return max(2, pool/8) } -// it only holds knots with work, so it can't grow with every knot seen +// only knots with work get entries, which keeps this from growing with every knot seen type notifySlots struct { mu sync.Mutex held map[string]int @@ -98,7 +97,7 @@ func (n *notifySlots) clock() time.Time { return time.Now() } -// a turned away knot counts as waiting, so it splits the pool with the rest +// a refused knot still goes in waiting, which shrinks everyone else's share func (n *notifySlots) acquire(host string, lim slotLimits) bool { host = canonicalKnotHost(host) n.mu.Lock() @@ -258,7 +257,7 @@ func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid sy 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 + // shutdown has to cancel this, or git keeps writing after the db closes if s.rootCtx != nil { defer context.AfterFunc(s.rootCtx, cancel)() } @@ -318,7 +317,7 @@ func (s *Spindle) runPush(ctx context.Context, k knotclient.Knot, repoDid syntax return s.handlePush(ctx, k, repo, push) } -// the db only has repos that passed ensureRepo, so only unknown ones read the knot +// a miss asks the knot because db only holds repos that passed ensureRepo 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 { @@ -346,7 +345,7 @@ func (s *Spindle) repoForPush(ctx context.Context, k knotclient.Knot, repoDid sy return repo, nil } -// the knot drops an event on any 4xx, so a refusal a retry can't fix gets one +// the knot drops an event on any 4xx, keep those for refusals a retry can't fix func refusedAsStatus(err error) error { if errors.Is(err, errRevRefused) || errors.Is(err, errTooManyWorkflowFiles) { return notifyStatus{http.StatusUnprocessableEntity, err} @@ -364,21 +363,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, so the 422 is final + // this 422 is final since a retry sends the same shas 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, so a banned user's push runs nothing even on an unbanned repo + // bans apply even when the repo isn't banned, because the knot names the pusher 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 } - // checked before the fetch, so a banned repo costs no git work + // the ban check runs 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 { @@ -429,7 +428,7 @@ func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Re return pipelineURI(k.Host, pipelineId), nil } -// the repo lock only covers these db reads and the webhook, the fetch can hang on a knot +// the lock covers only db reads and the webhook since a 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() @@ -439,7 +438,7 @@ 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 webhook fires on the first recording, even if the push fails later + // webhook fires on the first record, 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 { @@ -451,7 +450,7 @@ func (s *Spindle) admitPush(l *slog.Logger, k knotclient.Knot, repo *db.Repo, pu 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 fired above + // skip ci only skips the pipeline, the webhook already fired if kgit.HasSkipCIPushOption(push.PushOptions) { l.Info("push requested ci skip, skipping the event") return "", true, nil @@ -480,7 +479,7 @@ func isMaterializedRef(refname string) bool { return strings.HasPrefix(refname, "refs/heads/") || strings.HasPrefix(refname, "refs/tags/") } -// unknown changes run the workflows a paths filter would gate, since missing ci is worse +// unknown changes run the path-gated workflows too, 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 @@ -498,7 +497,7 @@ func changedFilesUnderBudget(ctx context.Context, l *slog.Logger, repoPath strin return admitted, true } -// the sync is shallow, so oldSha may be missing, and go-git can't read sha256 repos +// 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()) diff --git a/spindle/pulls.go b/spindle/pulls.go index 639f8ba0b..8f2dc1c74 100644 --- a/spindle/pulls.go +++ b/spindle/pulls.go @@ -21,7 +21,6 @@ import ( "tangled.org/core/workflow" ) -// read in batches so a big backlog never loads in one go const pullEventBatch = 16 // a pull request is a ticket with a target branch plus the patches that name it @@ -78,7 +77,7 @@ func (s *Spindle) applyPullRecord(ctx context.Context, rec *tapc.RecordEventData if err != nil || repo == nil { return err } - // a mark that lands during apply bumps the epoch, so the clear leaves its row alone + // a mark set during apply bumps the epoch, which makes the clear skip it epoch, err := s.db.RecordRetryEpoch(retry) if err != nil { return err @@ -89,7 +88,7 @@ func (s *Spindle) applyPullRecord(ctx context.Context, rec *tapc.RecordEventData if err := s.db.ClearRecordRetryAt(retry, epoch); err != nil { s.l.Error("clearing a record retry", "repo", rec.Did, "rkey", retry.Rkey, "err", err) } - s.startPullEvents(repo) + s.retries.jobs.kick(repo.RepoDid) return nil } @@ -146,7 +145,7 @@ func (s *Spindle) applyPatch(ctx context.Context, repo *db.Repo, rec *tapc.Recor return s.whileServed(repo, func() error { return s.db.AddPullPatch(repo.RepoDid, stored, ev) }) } -// the served check and the write share the repo lock, so a wipe can't let rows come back +// holding the repo lock means a wipe can't land between the check and the write func (s *Spindle) whileServed(repo *db.Repo, write func() error) error { defer s.lockRepo(repo.RepoDid).Unlock() if served, err := s.servedRepo(repo.RepoDid); err != nil || served == nil { @@ -201,17 +200,16 @@ func (s *Spindle) syncPullTicket(ctx context.Context, k knotclient.Knot, repo *d return &ticket, row, nil } -// a sync replays at most this many of the newest versions, so a heavily edited ticket stays bounded const maxPullVersionWalk = 64 -// events don't say what changed, so a close and reopen between syncs needs the versions +// events don't say what changed, and a close then reopen only shows up in the version walk func (s *Spindle) walkPullTicket(ctx context.Context, k knotclient.Knot, repo *db.Repo, rkey string, current *knotTicket, row *db.PullTicket) error { versions, err := s.knots.ListRecordVersions(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey) if errors.Is(err, knotclient.ErrNotFound) { return s.applyPullState(repo, rkey, current, row, "", current.Editor) } if err != nil { - // falling back to the newest state would skip the middle transitions for good + // using the newest state here would skip the middle transitions for good return fmt.Errorf("listing the versions of ticket %s/%s: %w", repo.RepoDid, rkey, err) } newest := len(versions) - 1 @@ -247,7 +245,6 @@ func (s *Spindle) walkPullTicket(ctx context.Context, k knotclient.Knot, repo *d return nil } -// an empty version means no history, so there is no version to record func (s *Spindle) applyPullState(repo *db.Repo, rkey string, at *knotTicket, row *db.PullTicket, version string, sender syntax.DID) error { state := *at.State event, moved := pullTransition(row.State, state) @@ -327,12 +324,6 @@ func newPullEvent(repo *db.Repo, rkey string, event models.WebhookEvent, ticket }, nil } -// one job per repo at a time keeps two runners off the same event -func (s *Spindle) startPullEvents(repo *db.Repo) { - s.retries.jobs.kick(repo.RepoDid) -} - -// a full notify budget leaves the rows due for the next pass func (s *Spindle) runDuePullEvents(ctx context.Context, did syntax.DID) { events, err := s.db.DuePullEvents(did, s.retries.clock(), pullEventBatch) if err != nil { @@ -347,7 +338,7 @@ func (s *Spindle) runDuePullEvents(ctx context.Context, did syntax.DID) { s.l.Error("looking up a repo with pull events", "repo", did, "err", err) return } - // events for an unserved repo would sit due forever, so they get dropped + // an unserved repo's events would stay due forever unless dropped if repo == nil { if err := s.db.DeletePullEventsByRepo(did); err != nil { s.l.Error("dropping the pull events of a repo no longer served", "repo", did, "err", err) @@ -355,19 +346,19 @@ func (s *Spindle) runDuePullEvents(ctx context.Context, did syntax.DID) { return } if s.runPullEvents(ctx, repo, events) && len(events) == pullEventBatch { - // back of the queue, so one long backlog can't hog a worker + // kick to the back of the queue, or one long backlog can hog a worker s.retries.jobs.kick(did) } } -// takes a notify slot per event, so a backlog doesn't pin the pool +// one notify slot per event, otherwise a backlog pins the pool func (s *Spindle) runPullEvents(ctx context.Context, repo *db.Repo, events []db.PullEvent) bool { for _, ev := range events { release, ok := s.acquireNotifySlot(repo.Knot) if !ok { return false } - // only a shutdown leaves the row untouched, a budget timeout counts as a failure + // only a shutdown skips the failure mark, a budget timeout still counts as one work, cancel := context.WithTimeout(ctx, notifyWorkBudget) err := s.runPullEvent(work, repo, ev) cancel() @@ -380,7 +371,6 @@ func (s *Spindle) runPullEvents(ctx context.Context, repo *db.Repo, events []db. return true } -// a banned author or sender means no webhook and no pipeline func (s *Spindle) bannedInPull(ev db.PullEvent) (bool, error) { for _, did := range []syntax.DID{ev.Author, ev.Sender} { if ban, err := s.db.GetBan(did); err != nil || ban != nil { diff --git a/spindle/schedule.go b/spindle/schedule.go index f7f2ba76e..b7b0909ac 100644 --- a/spindle/schedule.go +++ b/spindle/schedule.go @@ -83,13 +83,12 @@ type pipelineScheduler struct { byRepo map[string][]*scheduledWorkflow queue scheduleHeap - // refreshMu guards inflight and the store after a load, never held across a clone + // don't hold refreshMu across a clone refreshMu sync.Mutex inflight map[string]*refreshEntry dispatchMu sync.Mutex } -// the slot is held across the clone, and the entry goes once its last user leaves // one waiter at most, a newer one replaces it type refreshEntry struct { slot chan struct{} @@ -99,7 +98,7 @@ type refreshEntry struct { } type refreshWaiter struct { - // the removal count it asked at, the same count means it's covered + // removals seen when it asked, same count later means nothing was removed since at uint64 superseded chan struct{} } @@ -347,8 +346,8 @@ func (s *pipelineScheduler) workerLimit(work int) int { return limit } -// removals is read when the refresh asks, so a removal during the wait still drops the result -// a covered request skips the clone, the waiter loads for it +// read here, and a removal at any point after drops the result +// a covered request is served by the waiter that's already queued func (s *pipelineScheduler) beginRefresh(repoDid string) (e *refreshEntry, removals uint64, covered bool) { s.refreshMu.Lock() if s.inflight == nil { @@ -371,7 +370,7 @@ func (s *pipelineScheduler) beginRefresh(repoDid string) (e *refreshEntry, remov return e, removals, false default: } - // an old waiter would only drop its result after a removal, so it gives up its place + // only one waiter at a time, the newer one has the fresher removal count w := &refreshWaiter{at: removals, superseded: make(chan struct{})} if e.waiter != nil { close(e.waiter.superseded) @@ -462,7 +461,7 @@ func (s *pipelineScheduler) RefreshRepo(ctx context.Context, repo db.Repo) error return nil } -// doesn't wait on the clone, a clone can take minutes +// doesn't wait for an inflight clone, those can take minutes func (s *pipelineScheduler) RemoveRepo(ctx context.Context, repoDid string) error { s.refreshMu.Lock() defer s.refreshMu.Unlock() @@ -488,12 +487,7 @@ func (s *pipelineScheduler) replaceCachedSchedules(repoDid string, entries []sch if s.byRepo == nil { s.byRepo = make(map[string][]*scheduledWorkflow) } - for _, entry := range s.byRepo[repoDid] { - if entry.index >= 0 { - heap.Remove(&s.queue, entry.index) - } - } - delete(s.byRepo, repoDid) + s.dropCached(repoDid) after = after.UTC().Truncate(time.Minute) cached := make([]*scheduledWorkflow, 0, len(entries)) @@ -515,6 +509,11 @@ func (s *pipelineScheduler) replaceCachedSchedules(repoDid string, entries []sch func (s *pipelineScheduler) removeCachedSchedules(repoDid string) { s.mu.Lock() defer s.mu.Unlock() + s.dropCached(repoDid) +} + +// the caller holds mu +func (s *pipelineScheduler) dropCached(repoDid string) { for _, entry := range s.byRepo[repoDid] { if entry.index >= 0 { heap.Remove(&s.queue, entry.index) diff --git a/spindle/wipe.go b/spindle/wipe.go index 2428e0384..95c2a5a19 100644 --- a/spindle/wipe.go +++ b/spindle/wipe.go @@ -17,13 +17,13 @@ import ( "tangled.org/core/spindle/secrets" ) -// the wipes row outlives a restart, so the steps can pick up from it +// a restart can finish the wipe since the row is in the db func (s *Spindle) WipeRepo(ctx context.Context, repoDid syntax.DID, reason string) error { rows, err := s.db.ReposByDid(repoDid) if err != nil { return fmt.Errorf("list repo rows: %w", err) } - // force skips the turn check, so a refresh that read before this can't write over the wipe + // force skips the turn check, otherwise an older refresh could undo the wipe err = s.beginTurn(repoDid).force(func() error { return s.claimWipe(ctx, repoDid, rows, reason) }) if err != nil { return fmt.Errorf("claim %s: %w", repoDid, err) @@ -31,7 +31,7 @@ func (s *Spindle) WipeRepo(ctx context.Context, repoDid syntax.DID, reason strin return s.runWipe(ctx, repoDid) } -// cleanup is returned so the caller runs it after dropping the turn lock, not under it +// the caller runs cleanup after it drops the turn lock func (s *Spindle) claimWipeIfRows(ctx context.Context, repoDid syntax.DID, reason string) (cleanup func() error, err error) { rows, err := s.db.ReposByDid(repoDid) if err != nil { @@ -46,7 +46,6 @@ func (s *Spindle) claimWipeIfRows(ctx context.Context, repoDid syntax.DID, reaso return func() error { return s.runWipe(context.WithoutCancel(ctx), repoDid) }, nil } -// the repo's rows stay with the caller, so a failed revoke leaves the repo as it was func (s *Spindle) revokeRepoAccess(ctx context.Context, repoDid syntax.DID, rows []db.Repo) error { var errs []error collabs, err := s.db.ListCollaboratorsByRepoDid(repoDid) @@ -64,7 +63,7 @@ func (s *Spindle) revokeRepoAccess(ctx context.Context, repoDid syntax.DID, rows return errors.Join(errs...) } -// the wipes row written here is what stops the repo being served before cleanup ends +// the wipe row keeps a refresh from serving the repo during cleanup func (s *Spindle) claimWipe(ctx context.Context, repoDid syntax.DID, rows []db.Repo, reason string) error { if err := s.revokeRepoAccess(ctx, repoDid, rows); err != nil { return err @@ -80,10 +79,10 @@ func (s *Spindle) claimWipe(ctx context.Context, repoDid syntax.DID, rows []db.R return nil } -// run in this order, a failed step stays in the row for a retry +// order matters, a failed step stays in the row for a retry var wipeSteps = []string{"workflows", "leases", "artifacts", "jobs", "quota", "secrets", "webhooks", "pulls", "pipelines", "clone"} -// steps can rerun after a half finished one, so each only clears its own part +// each step has to be safe to repeat since a retry can rerun a half finished one func (s *Spindle) runWipe(ctx context.Context, repoDid syntax.DID) error { w, err := s.db.GetWipe(repoDid) if errors.Is(err, sql.ErrNoRows) { @@ -117,7 +116,7 @@ func (s *Spindle) runWipe(ctx context.Context, repoDid syntax.DID) error { return nil } -// no attempt cap, the wait just stops growing at repoRetryMax +// no attempt cap, backoff stops growing at repoRetryMax func wipeBackoff(attempts int) time.Duration { backoff := min(repoRetryBase<