From 4b5e366f5316aca3245a00ff12746b24278cba03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20Kupczy=C5=84ski?= Date: Thu, 15 Oct 2026 15:35:02 +0200 Subject: [PATCH] fix(engine): don't log routine concurrency index builds as errors Every scheduler restart or tenant rebalance makes the new owner wait out the previous holder's 30s outbox lease before the in-memory index builds. Run reported each of those ticks as an error, flooding alerts on rollouts. Builds under 60s now return ErrIndexBuildInProgress, logged at debug under a separate message; longer builds still log as errors. --- pkg/scheduling/v1/concurrency.go | 10 +++++-- pkg/scheduling/v1/concurrency/run_test.go | 35 +++++++++++++++++++++++ pkg/scheduling/v1/concurrency/strategy.go | 26 ++++++++++++++--- 3 files changed, 65 insertions(+), 6 deletions(-) diff --git a/pkg/scheduling/v1/concurrency.go b/pkg/scheduling/v1/concurrency.go index 41960c318d..55cdc7457d 100644 --- a/pkg/scheduling/v1/concurrency.go +++ b/pkg/scheduling/v1/concurrency.go @@ -2,6 +2,7 @@ package v1 import ( "context" + "errors" "fmt" "sync" "time" @@ -260,9 +261,14 @@ func (c *ConcurrencyManager) loopConcurrency(ctx context.Context) { results, err = c.repo.RunConcurrencyStrategy(ctx, c.tenantId, c.strategy) } c.releaseStrategyLocks() - if err != nil { + switch { + case errors.Is(err, concurrency.ErrIndexBuildInProgress): + // routine after a restart or rebalance; kept out of the "error running concurrency + // strategy" message so alerts on that message stay quiet at any log level + c.l.Debug().Ctx(ctx).Err(err).Msg("concurrency index still building, skipping run") + case err != nil: logger.ShutdownAware(ctx, c.l, err, zerolog.ErrorLevel).Ctx(ctx).Err(err).Msg("error running concurrency strategy") - } else if time.Since(start) > 100*time.Millisecond { + case time.Since(start) > 100*time.Millisecond: c.l.Warn().Ctx(ctx). Msgf("concurrency strategy %d took longer than 100ms (%s) to process %d items", c.strategy.ID, time.Since(start), len(results.Queued)) } diff --git a/pkg/scheduling/v1/concurrency/run_test.go b/pkg/scheduling/v1/concurrency/run_test.go index 7fa139e16c..ccb78757b2 100644 --- a/pkg/scheduling/v1/concurrency/run_test.go +++ b/pkg/scheduling/v1/concurrency/run_test.go @@ -167,3 +167,38 @@ func TestUpdateStrategyDuringOpenBatch(t *testing.T) { } } } + +// A build still inside the lease-handover window is routine and must be distinguishable from one +// that has run past stuckIndexBuildThreshold, which callers report as a real error. +func TestRunClassifiesUnfinishedIndexBuild(t *testing.T) { + restore := maxIndexBuildWaitPerRun + maxIndexBuildWaitPerRun = 10 * time.Millisecond + t.Cleanup(func() { maxIndexBuildWaitPerRun = restore }) + + cases := []struct { + name string + buildAge time.Duration + inProgress bool + }{ + {name: "within_lease_handover", buildAge: 0, inProgress: true}, + {name: "past_stuck_threshold", buildAge: 2 * stuckIndexBuildThreshold, inProgress: false}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c := newTestStrategy(&mockConcurrencyRepo{}, 5) + c.buildStartedAt = time.Now().Add(-tc.buildAge) + + res, err := c.Run(context.Background()) + if err == nil { + t.Fatalf("expected an error while the index is still building") + } + if res != nil { + t.Fatalf("expected no results while the index is still building, got %+v", res) + } + if got := errors.Is(err, ErrIndexBuildInProgress); got != tc.inProgress { + t.Fatalf("errors.Is(err, ErrIndexBuildInProgress) = %v, want %v (err: %v)", got, tc.inProgress, err) + } + }) + } +} diff --git a/pkg/scheduling/v1/concurrency/strategy.go b/pkg/scheduling/v1/concurrency/strategy.go index 3ff14081a0..23de05a90b 100644 --- a/pkg/scheduling/v1/concurrency/strategy.go +++ b/pkg/scheduling/v1/concurrency/strategy.go @@ -3,6 +3,7 @@ package concurrency import ( "context" "encoding/json" + "errors" "fmt" "runtime" "sync" @@ -30,15 +31,26 @@ const ( // slowTopicAcquireThreshold is how long acquiring the outbox topic lease may take before // buildIndex logs it: the acquire blocks silently until the previous holder's lease lapses. slowTopicAcquireThreshold = 10 * time.Second + + // stuckIndexBuildThreshold is how long the initial index build may run before Run reports it + // as stuck. Every scheduler restart or tenant rebalance makes the new owner wait out the + // previous holder's exclusive outbox lease (30s as of pgoutbox v0.4.0), so builds shorter + // than this are routine. + stuckIndexBuildThreshold = 60 * time.Second ) // maxIndexBuildWaitPerRun bounds how long a single Run waits for the initial index build. // The caller holds the strategy's in-memory lock for the whole Run, so a build that never -// finishes (e.g. the outbox topic lease is stuck on another instance) must surface as a -// repeated, visible error rather than one goroutine blocked forever holding that lock. -// A var rather than a const so tests can shorten it. +// finishes (e.g. the outbox topic lease is stuck on another instance) must make Run return +// rather than block forever holding that lock. Once the build exceeds stuckIndexBuildThreshold, +// every Run reports it as an error. A var rather than a const so tests can shorten it. var maxIndexBuildWaitPerRun = 15 * time.Second +// ErrIndexBuildInProgress is returned by Run when the initial index build has not finished but +// has been running for less than stuckIndexBuildThreshold. It is routine rather than a failure: +// the run is skipped and a later Run proceeds once the build completes. +var ErrIndexBuildInProgress = errors.New("concurrency index is still building") + type ConcurrencyStrategy struct { outbox pgoutbox.Outbox repo repository.ConcurrencyRepository @@ -261,7 +273,13 @@ func (c *ConcurrencyStrategy) Run(ctx context.Context) (*repository.RunConcurren case <-ctx.Done(): return nil, fmt.Errorf("timed out waiting for concurrency index to finish building for topic %s: %w", c.topic, ctx.Err()) case <-buildWaitTimer.C: - return nil, fmt.Errorf("concurrency index for topic %s is still building after %s; skipping this run", c.topic, time.Since(c.buildStartedAt).Round(time.Second)) + elapsed := time.Since(c.buildStartedAt) + + if elapsed < stuckIndexBuildThreshold { + return nil, fmt.Errorf("%w for topic %s after %s; skipping this run", ErrIndexBuildInProgress, c.topic, elapsed.Round(time.Second)) + } + + return nil, fmt.Errorf("concurrency index for topic %s is still building after %s, longer than the %ds stuck threshold; skipping this run", c.topic, elapsed.Round(time.Second), int(stuckIndexBuildThreshold.Seconds())) } // discard any results left over from a previous aborted Run before we start draining