fix(webhook): reject new syncs during shutdown

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

Author: tae2089

internal/webhook/handler.go

@@ -161,6 +161,10 @@ func (h *WebhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			http.Error(w, "sync queue full", http.StatusTooManyRequests)
 			return
 		}
+		if err == ErrSyncQueueShuttingDown {
+			http.Error(w, "sync queue shutting down", http.StatusServiceUnavailable)
+			return
+		}
 		http.Error(w, "sync dispatch failed", http.StatusInternalServerError)
 		return
 	}

internal/webhook/handler_test.go

@@ -517,6 +517,31 @@ func TestWebhookHandler_ReturnsTooManyRequestsWhenSyncQueueFull(t *testing.T) {
 	}
 }
 
+func TestWebhookHandler_ReturnsServiceUnavailableWhenSyncQueueShuttingDown(t *testing.T) {
+	secret := []byte("test-secret")
+	al := NewRepoFilter([]string{"org/*"})
+	h := NewWebhookHandlerWithConfig(WebhookHandlerConfig{
+		Secret:        secret,
+		Filter:        al,
+		CloneBaseURLs: []string{"https://github.com"},
+		OnSync: func(context.Context, string, string, string) error {
+			return ErrSyncQueueShuttingDown
+		},
+	})
+
+	payload := makePushEvent("refs/heads/main", "org/svc", "https://github.com/org/svc.git")
+	req := httptest.NewRequest(http.MethodPost, "/webhook", bytes.NewReader(payload))
+	req.Header.Set("X-Hub-Signature-256", signPayload(secret, payload))
+	req.Header.Set("X-GitHub-Event", "push")
+
+	rr := httptest.NewRecorder()
+	h.ServeHTTP(rr, req)
+
+	if rr.Code != http.StatusServiceUnavailable {
+		t.Fatalf("status = %d, want %d", rr.Code, http.StatusServiceUnavailable)
+	}
+}
+
 func TestWebhookHandler_PropagatesTraceparentToSyncContext(t *testing.T) {
 	secret := []byte("test-secret")
 	al := NewRepoFilter([]string{"org/*"})

internal/webhook/syncqueue.go

@@ -15,7 +15,10 @@ import (
 	"github.com/tae2089/code-context-graph/internal/obs"
 )
 
-var ErrSyncQueueFull = errors.New("sync queue full")
+var (
+	ErrSyncQueueFull         = errors.New("sync queue full")
+	ErrSyncQueueShuttingDown = errors.New("sync queue shutting down")
+)
 
 // @intent mark sync failures that should stop retry backoff immediately.
 type nonRetryableError struct {
@@ -98,6 +101,7 @@ type SyncQueue struct {
 	ctx             context.Context
 	handler         SyncHandlerFunc
 	retryConfig     RetryConfig
+	shutdownTimeout time.Duration
 	maxTrackedRepos int
 	mu              sync.Mutex
 	queue           []string
@@ -164,6 +168,7 @@ func NewSyncQueueWithContext(ctx context.Context, workers int, handler SyncHandl
 // @intent configure queue retry policy and memory bounds when constructing a SyncQueue.
 type QueueConfig struct {
 	RetryConfig
+	ShutdownTimeout time.Duration
 	MaxTrackedRepos int
 }
 
@@ -179,21 +184,25 @@ func NewSyncQueueWithOptions(ctx context.Context, workers int, handler SyncHandl
 // @intent coalesce bursty webhook pushes per repository while still allowing different repos to sync concurrently.
 // @param cfg controls retry behavior and how many repositories may be tracked at once.
 // @sideEffect starts worker goroutines and allocates queue state maps.
-// @ensures non-positive MaxAttempts becomes 1 and non-positive MaxTrackedRepos becomes 1024.
+// @ensures non-positive MaxAttempts becomes 1, non-positive ShutdownTimeout becomes 30s, and non-positive MaxTrackedRepos becomes 1024.
 func NewSyncQueueWithConfig(ctx context.Context, workers int, handler SyncHandlerFunc, cfg QueueConfig) *SyncQueue {
 	if ctx == nil {
 		ctx = context.Background()
 	}
 	if cfg.MaxAttempts <= 0 {
 		cfg.MaxAttempts = 1
 	}
+	if cfg.ShutdownTimeout <= 0 {
+		cfg.ShutdownTimeout = 30 * time.Second
+	}
 	if cfg.MaxTrackedRepos <= 0 {
 		cfg.MaxTrackedRepos = 1024
 	}
 	q := &SyncQueue{
 		ctx:             ctx,
 		handler:         handler,
 		retryConfig:     cfg.RetryConfig,
+		shutdownTimeout: cfg.ShutdownTimeout,
 		maxTrackedRepos: cfg.MaxTrackedRepos,
 		dirty:           make(map[string]bool),
 		processing:      make(map[string]bool),
@@ -216,7 +225,7 @@ func NewSyncQueueWithConfig(ctx context.Context, workers int, handler SyncHandle
 // @intent collapse repeated push events into one queued sync while preserving the newest branch and clone data.
 // @mutates SyncQueue.queue, SyncQueue.dirty, SyncQueue.payloads.
 // @param repoFullName identifies the deduplication key for queueing.
-// @return returns ErrSyncQueueFull when a new repository would exceed the bounded tracked-repo set.
+// @return returns ErrSyncQueueFull when a new repository would exceed the bounded tracked-repo set and ErrSyncQueueShuttingDown after graceful shutdown begins.
 // @domainRule repeated events for the same repository replace payload data instead of enqueueing duplicate work.
 func (q *SyncQueue) Add(ctx context.Context, repoFullName, cloneURL, branch string) error {
 	if ctx != nil {
@@ -230,7 +239,7 @@ func (q *SyncQueue) Add(ctx context.Context, repoFullName, cloneURL, branch stri
 	defer q.mu.Unlock()
 
 	if q.shutdown {
-		return nil
+		return ErrSyncQueueShuttingDown
 	}
 
 	if ctx == nil {
@@ -240,7 +249,7 @@ func (q *SyncQueue) Add(ctx context.Context, repoFullName, cloneURL, branch stri
 		q.queueFullTotal++
 		return ErrSyncQueueFull
 	}
-	q.payloads[repoFullName] = syncPayload{ctx: ctx, repoFullName: repoFullName, cloneURL: cloneURL, branch: branch}
+	q.payloads[repoFullName] = syncPayload{ctx: context.WithoutCancel(ctx), repoFullName: repoFullName, cloneURL: cloneURL, branch: branch}
 
 	now := time.Now()
 	if q.dirty[repoFullName] {
@@ -279,9 +288,10 @@ func (q *SyncQueue) Shutdown() {
 
 	select {
 	case <-done:
-	case <-time.After(30 * time.Second):
-		slog.Error("sync queue shutdown timed out after 30s, abandoning workers")
-		q.recordFailure("shutdown", "", errors.New("sync queue shutdown timed out after 30s"))
+	case <-time.After(q.shutdownTimeout):
+		err := fmt.Errorf("sync queue shutdown timed out after %s", q.shutdownTimeout)
+		slog.Error(err.Error())
+		q.recordFailure("shutdown", "", err)
 	}
 }
 

internal/webhook/syncqueue_test.go

@@ -405,28 +405,42 @@ func TestSyncQueue_AddHonorsPerCallContextCancel(t *testing.T) {
 	}
 }
 
-func TestSyncQueue_AddPerCallTimeoutPropagates(t *testing.T) {
+func TestSyncQueue_AddDetachesPerCallTimeoutAfterEnqueue(t *testing.T) {
+	handlerStarted := make(chan struct{}, 1)
 	handlerDone := make(chan error, 1)
+	release := make(chan struct{})
 	handler := func(ctx context.Context, repoFullName, cloneURL, branch string) error {
-		<-ctx.Done()
+		handlerStarted <- struct{}{}
+		<-release
 		handlerDone <- ctx.Err()
-		return ctx.Err()
+		return nil
 	}
 
 	q := NewSyncQueue(1, handler)
 	defer q.Shutdown()
 
 	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
 	defer cancel()
-	q.Add(ctx, "org/svc", "url", "main")
+	if err := q.Add(ctx, "org/svc", "url", "main"); err != nil {
+		t.Fatalf("Add returned error: %v", err)
+	}
+
+	select {
+	case <-handlerStarted:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for handler start")
+	}
+
+	time.Sleep(100 * time.Millisecond)
+	close(release)
 
 	select {
 	case err := <-handlerDone:
-		if !errors.Is(err, context.DeadlineExceeded) {
-			t.Fatalf("expected deadline exceeded, got %v", err)
+		if err != nil {
+			t.Fatalf("expected detached handler context, got %v", err)
 		}
 	case <-time.After(2 * time.Second):
-		t.Fatal("timed out waiting for handler")
+		t.Fatal("timed out waiting for handler completion")
 	}
 }
 
@@ -546,6 +560,54 @@ func TestSyncQueue_AddAllowsExistingRepoUpdateWhenAtCapacity(t *testing.T) {
 	}
 }
 
+func TestSyncQueue_AddRejectsAfterShutdownStarts(t *testing.T) {
+	q := NewSyncQueueWithConfig(context.Background(), 0, func(context.Context, string, string, string) error {
+		return nil
+	}, QueueConfig{
+		RetryConfig:     RetryConfig{MaxAttempts: 1, BaseDelay: time.Millisecond, MaxDelay: time.Millisecond},
+		ShutdownTimeout: time.Second,
+		MaxTrackedRepos: 1,
+	})
+	defer q.Shutdown()
+
+	q.Shutdown()
+
+	err := q.Add(context.Background(), "org/svc", "url", "main")
+	if !errors.Is(err, ErrSyncQueueShuttingDown) {
+		t.Fatalf("Add error = %v, want %v", err, ErrSyncQueueShuttingDown)
+	}
+}
+
+func TestSyncQueue_ShutdownUsesConfiguredTimeout(t *testing.T) {
+	release := make(chan struct{})
+	q := NewSyncQueueWithConfig(context.Background(), 1, func(context.Context, string, string, string) error {
+		<-release
+		return nil
+	}, QueueConfig{
+		RetryConfig:     RetryConfig{MaxAttempts: 1, BaseDelay: time.Millisecond, MaxDelay: time.Millisecond},
+		ShutdownTimeout: 30 * time.Millisecond,
+		MaxTrackedRepos: 1,
+	})
+
+	if err := q.Add(context.Background(), "org/svc", "url", "main"); err != nil {
+		t.Fatalf("Add returned error: %v", err)
+	}
+	time.Sleep(10 * time.Millisecond)
+
+	start := time.Now()
+	q.Shutdown()
+	elapsed := time.Since(start)
+	close(release)
+
+	if elapsed > 250*time.Millisecond {
+		t.Fatalf("Shutdown took %s, want close to configured timeout", elapsed)
+	}
+	stats := q.Stats()
+	if stats.FailureTotal == 0 {
+		t.Fatalf("expected shutdown timeout failure to be recorded, got %+v", stats)
+	}
+}
+
 func TestSyncQueueStats_ReportsQueueFull(t *testing.T) {
 	q := NewSyncQueueWithConfig(context.Background(), 0, func(context.Context, string, string, string) error {
 		return nil
@@ -899,8 +961,9 @@ func TestSyncQueueStats_RecentRepos_IncludesQueuedAndProcessingBeforeFirstOutcom
 func TestSyncQueueStats_ReportsAgesAndLastSuccess(t *testing.T) {
 	release := make(chan struct{})
 	started := make(chan struct{})
+	var once sync.Once
 	q := NewSyncQueueWithConfig(context.Background(), 1, func(context.Context, string, string, string) error {
-		close(started)
+		once.Do(func() { close(started) })
 		<-release
 		return nil
 	}, QueueConfig{