fix(engine): don't log routine concurrency index builds as errors

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))
 		}

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)
+			}
+		})
+	}
+}

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