AdaAibaby · Nov 12, 2026
@@ -100,6 +100,12 @@ func (s *Storage) healExpirationIndex(ctx context.Context) (int, teamWalk, error
 	}
 	walk.complete = skipped == 0
 
+	// Reuse the scan's own skipped-team count (per-team failures silenced by
+	// forEachSandboxBatch) as the teams_skipped signal — no extra bookkeeping.
+	if skipped > 0 {
+		s.scanMet.teamsSkipped.Add(ctx, int64(skipped))
+	}
+
 	return healed, walk, nil
 }
 
@@ -554,7 +560,7 @@ func (s *Storage) readRecordOf(ctx context.Context, key string, ref recordRef) (
 	if err != nil {
 		return sandboxtypes.Sandbox{}, false, fmt.Errorf("GET: %w", err)
 	}
-	sbx, ok := decodeRecord(ctx, readRef{ref: ref, team: team}, data)
+	sbx, ok := s.decodeRecord(ctx, readRef{ref: ref, team: team}, data)
 
 	return sbx, ok, nil
 }
@@ -166,3 +166,23 @@ func drainLockNotification(ch <-chan struct{}) {
 	default:
 	}
 }
+
+// obtainLock wraps storageLocker.Obtain with wait-duration and timeout metrics.
+// On success it records the total wait time; on context expiry it increments the
+// timeout counter. All other errors pass through unrecorded.
+func (s *Storage) obtainLock(ctx context.Context, lockKey string, timeout time.Duration) (*storageLock, error) {
+	start := time.Now()
+
+	lock, err := s.locker.Obtain(ctx, lockKey, timeout)
+	if err != nil {
+		if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+			s.lockMet.timeouts.Add(context.WithoutCancel(ctx), 1)
+		}
+
+		return nil, err
+	}
+
+	s.lockMet.waitDuration.Record(ctx, time.Since(start).Milliseconds())
+
+	return lock, nil
+}
@@ -54,7 +54,10 @@ type Storage struct {
 
 	reconcileSlots *semaphore.Weighted
 
-	metrics expirationIndexMetrics
+	metrics       expirationIndexMetrics
+	lockMet       lockMetrics
+	transitionMet transitionMetrics
+	scanMet       scanMetrics
 	// writeAnomalies counts writes that gave up after their retries, by kind.
 	writeAnomalies metric.Int64Counter
 	// listGets records the GETs one bulk record read issued.
@@ -84,6 +87,28 @@ type expirationIndexMetrics struct {
 	sweptOrphan        metric.MeasurementOption
 	sweptDeadExecution metric.MeasurementOption
 	sweptInvalid       metric.MeasurementOption
+
+	// teamIndexSize records each team's sandbox index SET cardinality, sampled
+	// where TeamsWithSandboxCount already pipelines SCARD over every team.
+	teamIndexSize metric.Int64Histogram
+}
+
+type lockMetrics struct {
+	waitDuration metric.Int64Histogram
+	timeouts     metric.Int64Counter
+}
+
+type transitionMetrics struct {
+	callbackFailures metric.Int64Counter
+	duration         metric.Int64Histogram
+	joined           metric.Int64Counter
+	fallbackPolls    metric.Int64Counter
+	mismatches       metric.Int64Counter
+}
+
+type scanMetrics struct {
+	teamsSkipped   metric.Int64Counter
+	corruptRecords metric.Int64Counter
 }
 
 const sweptReasonAttr = "reason"
@@ -109,6 +134,11 @@ func newExpirationIndexMetrics(meter metric.Meter) (expirationIndexMetrics, erro
 		return expirationIndexMetrics{}, fmt.Errorf("expiration index swept counter: %w", err)
 	}
 
+	teamIndexSize, err := telemetry.GetHistogram(meter, telemetry.ApiRedisStorageTeamIndexSize)
+	if err != nil {
+		return expirationIndexMetrics{}, fmt.Errorf("team index size histogram: %w", err)
+	}
+
 	return expirationIndexMetrics{
 		indexHealed:        healed,
 		indexRescored:      rescored,
@@ -117,9 +147,73 @@ func newExpirationIndexMetrics(meter metric.Meter) (expirationIndexMetrics, erro
 		sweptOrphan:        metric.WithAttributeSet(attribute.NewSet(attribute.String(sweptReasonAttr, "orphan"))),
 		sweptDeadExecution: metric.WithAttributeSet(attribute.NewSet(attribute.String(sweptReasonAttr, "dead_execution"))),
 		sweptInvalid:       metric.WithAttributeSet(attribute.NewSet(attribute.String(sweptReasonAttr, "invalid"))),
+		teamIndexSize:      teamIndexSize,
 	}, nil
 }
 
+func newLockMetrics(meter metric.Meter) (lockMetrics, error) {
+	waitDur, err := telemetry.GetHistogram(meter, telemetry.ApiRedisStorageLockWaitDuration)
+	if err != nil {
+		return lockMetrics{}, fmt.Errorf("lock wait duration histogram: %w", err)
+	}
+
+	timeouts, err := telemetry.GetCounter(meter, telemetry.ApiRedisStorageLockTimeouts)
+	if err != nil {
+		return lockMetrics{}, fmt.Errorf("lock timeouts counter: %w", err)
+	}
+
+	return lockMetrics{waitDuration: waitDur, timeouts: timeouts}, nil
+}
+
+func newTransitionMetrics(meter metric.Meter) (transitionMetrics, error) {
+	cbFails, err := telemetry.GetCounter(meter, telemetry.ApiRedisStorageTransitionCallbackFailures)
+	if err != nil {
+		return transitionMetrics{}, fmt.Errorf("transition callback failures counter: %w", err)
+	}
+
+	dur, err := telemetry.GetHistogram(meter, telemetry.ApiRedisStorageTransitionDuration)
+	if err != nil {
+		return transitionMetrics{}, fmt.Errorf("transition duration histogram: %w", err)
+	}
+
+	joined, err := telemetry.GetCounter(meter, telemetry.ApiRedisStorageTransitionJoined)
+	if err != nil {
+		return transitionMetrics{}, fmt.Errorf("transition joined counter: %w", err)
+	}
+
+	fallbacks, err := telemetry.GetCounter(meter, telemetry.ApiRedisStorageTransitionFallbackPolls)
+	if err != nil {
+		return transitionMetrics{}, fmt.Errorf("transition fallback polls counter: %w", err)
+	}
+
+	mismatches, err := telemetry.GetCounter(meter, telemetry.ApiRedisStorageTransitionExecutionMismatches)
+	if err != nil {
+		return transitionMetrics{}, fmt.Errorf("transition execution mismatches counter: %w", err)
+	}
+
+	return transitionMetrics{
+		callbackFailures: cbFails,
+		duration:         dur,
+		joined:           joined,
+		fallbackPolls:    fallbacks,
+		mismatches:       mismatches,
+	}, nil
+}
+
+func newScanMetrics(meter metric.Meter) (scanMetrics, error) {
+	skipped, err := telemetry.GetCounter(meter, telemetry.ApiRedisStorageScanTeamsSkipped)
+	if err != nil {
+		return scanMetrics{}, fmt.Errorf("scan teams skipped counter: %w", err)
+	}
+
+	corrupt, err := telemetry.GetCounter(meter, telemetry.ApiRedisStorageScanCorruptRecords)
+	if err != nil {
+		return scanMetrics{}, fmt.Errorf("scan corrupt records counter: %w", err)
+	}
+
+	return scanMetrics{teamsSkipped: skipped, corruptRecords: corrupt}, nil
+}
+
 // Write anomaly kinds, the metric's kind attribute: a record left after its
 // index entry was removed, when its delete kept failing,
 // which the healer deletes once it is stale; a record written without its
@@ -157,6 +251,21 @@ func NewStorage(
 		return nil, fmt.Errorf("failed to create expiration index metrics: %w", err)
 	}
 
+	lockMet, err := newLockMetrics(meter)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create lock metrics: %w", err)
+	}
+
+	transitionMet, err := newTransitionMetrics(meter)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create transition metrics: %w", err)
+	}
+
+	scanMet, err := newScanMetrics(meter)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create scan metrics: %w", err)
+	}
+
 	// A misnamed instrument panics at construction, as the package-level
 	// instruments do.
 	s := &Storage{
@@ -166,6 +275,9 @@ func NewStorage(
 		publisher:      pub,
 		reconcileSlots: semaphore.NewWeighted(maxConcurrentReconciles),
 		metrics:        metrics,
+		lockMet:        lockMet,
+		transitionMet:  transitionMet,
+		scanMet:        scanMet,
 		writeAnomalies: utils.Must(telemetry.GetCounter(meter, telemetry.ApiRedisStorageWriteAnomalies)),
 		listGets:       utils.Must(telemetry.GetHistogram(meter, telemetry.ApiRedisStorageListGets)),
 		indexHeal:      utils.Must(telemetry.GetCounter(meter, telemetry.ApiRedisStorageIndexHeal)),

@@ -586,6 +586,9 @@ func (s *Storage) TeamsWithSandboxCount(ctx context.Context) (map[uuid.UUID]int6
 	for _, e := range entries {
 		if count := e.cmd.Val(); count > 0 {
 			teams[e.id] = count
+			// Reuse this already-pipelined SCARD as the team-index-size sample
+			// instead of a second full SCARD pass over globalTeamsSet.
+			s.metrics.teamIndexSize.Record(ctx, count)
 		} else if int64(e.score) < cutoff {
 			// Only prune if the entry is old enough — a fresh score means
 			// an Add happened recently and SCARD==0 may be a transient race.

@@ -128,7 +128,7 @@ func (s *Storage) getRecords(ctx context.Context, reads []readRef, found map[rec
 
 			return nil, fmt.Errorf("failed to read the record of sandbox %s: %w", r.ref.SandboxID, err)
 		}
-		if sbx, ok := decodeRecord(ctx, r, data); ok {
+		if sbx, ok := s.decodeRecord(ctx, r, data); ok {
 			found[r.ref] = sbx
 		}
 	}
@@ -142,9 +142,10 @@ func (s *Storage) getRecords(ctx context.Context, reads []readRef, found map[rec
 // or a record naming another sandbox than its key, from whose ID the callers
 // would build keys, is logged and left out. Teams are compared as UUIDs,
 // never as strings.
-func decodeRecord(ctx context.Context, r readRef, data []byte) (sandboxtypes.Sandbox, bool) {
+func (s *Storage) decodeRecord(ctx context.Context, r readRef, data []byte) (sandboxtypes.Sandbox, bool) {
 	var sbx sandboxtypes.Sandbox
 	if err := json.Unmarshal(data, &sbx); err != nil {
+		s.scanMet.corruptRecords.Add(ctx, 1)
 		// Warn, not Error: every scan pass re-reads the same record, so a
 		// single corrupt one would otherwise page on a loop.
 		logger.L().Warn(ctx, "Skipping sandbox record that does not decode",

@@ -37,6 +37,8 @@ import (
 // The callback is critical: it deletes the transition key
 // and sets the result value with short TTL to notify waiters of the outcome.
 func (s *Storage) StartRemoving(ctx context.Context, teamID uuid.UUID, sandboxID string, opts sandboxtypes.RemoveOpts) (sandboxtypes.StateTransition, bool, func(context.Context, error), error) {
+	transitionStart := time.Now()
+
 	if err := s.refuseOtherTeam(ctx, teamID.String(), sandboxID); err != nil {
 		return sandboxtypes.StateTransition{}, false, nil, err
 	}
@@ -45,7 +47,7 @@ func (s *Storage) StartRemoving(ctx context.Context, teamID uuid.UUID, sandboxID
 	lockKey := redis_utils.GetLockKey(key)
 
 	// Acquire distributed lock
-	lock, err := s.locker.Obtain(ctx, lockKey, lockTimeout)
+	lock, err := s.obtainLock(ctx, lockKey, lockTimeout)
 	if err != nil {
 		return sandboxtypes.StateTransition{}, false, nil, fmt.Errorf("failed to obtain lock: %w", err)
 	}
@@ -78,6 +80,8 @@ func (s *Storage) StartRemoving(ctx context.Context, teamID uuid.UUID, sandboxID
 	// equivalent check inside startTransitionScript, which is atomic with that
 	// write.
 	if opts.ExpectExecutionID != "" && sbx.ExecutionID != opts.ExpectExecutionID {
+		s.transitionMet.mismatches.Add(ctx, 1)
+
 		return transition, false, nil, fmt.Errorf(
 			"sandbox %q is execution %q, expected %q: %w",
 			sandboxID, sbx.ExecutionID, opts.ExpectExecutionID, sandboxtypes.ErrExecutionMismatch,
@@ -174,6 +178,8 @@ func (s *Storage) StartRemoving(ctx context.Context, teamID uuid.UUID, sandboxID
 		// The pin no longer holds: the incarnation was replaced since the check
 		// above, past a lock that expired; the script refused rather than
 		// overwrite whatever is there now, and nothing has been written.
+		s.transitionMet.mismatches.Add(ctx, 1)
+
 		return transition, false, nil, fmt.Errorf(
 			"sandbox %q is no longer execution %q: %w",
 			sandboxID, opts.ExpectExecutionID, sandboxtypes.ErrExecutionMismatch,
@@ -185,16 +191,18 @@ func (s *Storage) StartRemoving(ctx context.Context, teamID uuid.UUID, sandboxID
 	transition.Sandbox = updated
 	transition.OriginalEndTime = &sbx.EndTime
 
-	return transition, false, s.createCallback(teamID, sandboxID, transitionKey, resultKey, transitionID, opts.Action), nil
+	return transition, false, s.createCallback(teamID, sandboxID, transitionKey, resultKey, transitionID, opts.Action, transitionStart), nil
 }
 
 // createCallback returns a callback function for completing a transition.
 // A successful transient action restores Running unless the caller passes
 // ErrSkipRestore. Completing a transient action with an error also skips
 // restoration and lets waiters proceed. Waiters must read the resulting state.
 // Automatic restore failures and non-transient errors are published to waiters.
-func (s *Storage) createCallback(teamID uuid.UUID, sandboxID, transitionKey, resultKey, transitionID string, stateAction sandboxtypes.StateAction) func(context.Context, error) {
+func (s *Storage) createCallback(teamID uuid.UUID, sandboxID, transitionKey, resultKey, transitionID string, stateAction sandboxtypes.StateAction, start time.Time) func(context.Context, error) {
 	return func(cbCtx context.Context, cbErr error) {
+		s.transitionMet.duration.Record(cbCtx, time.Since(start).Milliseconds())
+
 		skipRestore := errors.Is(cbErr, sandboxtypes.ErrSkipRestore)
 		if skipRestore {
 			cbErr = nil
@@ -229,6 +237,7 @@ func (s *Storage) createCallback(teamID uuid.UUID, sandboxID, transitionKey, res
 		err := RunScript(cbCtx, s.redisClient, "finish_transition", finishTransitionScript, []string{transitionKey, resultKey, refusalClassKey(resultKey)},
 			transitionID, resultValue, int(transitionResultKeyTTL.Seconds()), refusalClass).Err()
 		if err != nil {
+			s.transitionMet.callbackFailures.Add(context.WithoutCancel(cbCtx), 1)
 			logger.L().Warn(cbCtx, "Failed to finish state transition", logger.WithSandboxID(sandboxID), zap.String("transitionID", transitionID), zap.Error(err))
 		}
 
@@ -422,6 +431,8 @@ func (s *Storage) waitForTransition(ctx context.Context, sandboxID, transitionID
 		case <-ch:
 			return s.checkTransitionResult(ctx, resultKey)
 		case <-ticker.C:
+			s.transitionMet.fallbackPolls.Add(ctx, 1)
+
 			// Fallback poll: check whether the transition key is still present.
 			currentID, err := s.redisClient.Get(ctx, transitionKey).Result()
 			if errors.Is(err, redis.Nil) || currentID != transitionID {
@@ -504,6 +515,7 @@ func (s *Storage) handleExistingTransition(
 		// The caller inherits the in-flight transition's result without
 		// doing the work itself: this is a joiner.
 		joined.Mark(ctx)
+		s.transitionMet.joined.Add(ctx, 1)
 
 		logger.L().Debug(ctx, "State transition already in progress to the same state, waiting",
 			logger.WithSandboxID(sbx.SandboxID),