Merge pull request #117 from flashcatcloud/fix/incident-verify-concurrent

perf(incident): verify comment write-back concurrently

Author: ysyneu

internal/cli/command_test.go

@@ -10,7 +10,10 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/flashcatcloud/go-flashduty"
 	"github.com/spf13/cobra"
@@ -820,14 +823,17 @@ func TestCommandIncidentCommentFailsWhenNoCommentEntryFound(t *testing.T) {
 func TestCommandIncidentCommentChecksEntireBatchNotJustFirstFailure(t *testing.T) {
 	saveAndResetGlobals(t)
 	stub := newGFStub(t)
+	var mu sync.Mutex
 	feedCallsByIncident := map[string]int{}
 	stub.dataForPath = func(path string, body map[string]any) any {
 		switch path {
 		case "/incident/comment":
 			return map[string]any{}
 		case "/incident/feed":
 			incID, _ := body["incident_id"].(string)
+			mu.Lock()
 			feedCallsByIncident[incID]++
+			mu.Unlock()
 			switch incID {
 			case "inc-2":
 				return map[string]any{"items": []any{
@@ -984,6 +990,163 @@ func TestCommandIncidentCommentVerifiesAcrossFeedPages(t *testing.T) {
 	}
 }
 
+// TestCommandIncidentCommentVerifyOrderStableDespiteCompletionOrder guards
+// that concurrent verification's problem list comes out in ctx.Args order,
+// never completion order. The stub sleeps longer for earlier-indexed
+// incidents than later ones, so completion order is guaranteed to be the
+// exact reverse of argument order; a test that let requests finish in
+// whatever order they naturally would (e.g. all equally fast) could pass
+// even with results appended in completion order, since ctx.Args order and
+// completion order would coincidentally match. Forcing the inversion is what
+// makes this a real test of the ordering guarantee rather than of luck.
+func TestCommandIncidentCommentVerifyOrderStableDespiteCompletionOrder(t *testing.T) {
+	saveAndResetGlobals(t)
+	stub := newGFStub(t)
+
+	const n = 4
+	ids := make([]string, n)
+	delay := make(map[string]time.Duration, n)
+	for i := range ids {
+		ids[i] = fmt.Sprintf("inc-%d", i+1)
+		// inc-1 sleeps longest, inc-n shortest: completion order is guaranteed
+		// to be inc-n, ..., inc-1 — the reverse of ids' (== ctx.Args') order.
+		delay[ids[i]] = time.Duration(n-i) * 15 * time.Millisecond
+	}
+	stub.dataForPath = func(path string, body map[string]any) any {
+		switch path {
+		case "/incident/comment":
+			return map[string]any{}
+		case "/incident/feed":
+			incID, _ := body["incident_id"].(string)
+			time.Sleep(delay[incID])
+			// Empty timeline: every incident fails verification, so every one
+			// of them appears in the problem list this test inspects.
+			return map[string]any{"items": []any{}}
+		default:
+			return map[string]any{}
+		}
+	}
+	commentFile := writeCommentFile(t, "the real comment")
+
+	args := append([]string{"incident", "comment"}, ids...)
+	args = append(args, "--comment-file", commentFile)
+	_, err := execCommand(args...)
+	if err == nil {
+		t.Fatal("[verify-order] expected a non-zero exit, got nil error")
+	}
+
+	lastIdx := -1
+	for _, id := range ids {
+		idx := strings.Index(err.Error(), "incident "+id+":")
+		if idx == -1 {
+			t.Fatalf("[verify-order] expected %q named in the error: %v", id, err)
+		}
+		if idx <= lastIdx {
+			t.Fatalf("[verify-order] problems are not in ctx.Args order (completion order leaked through) in: %v", err)
+		}
+		lastIdx = idx
+	}
+}
+
+// incidentFeedConcurrencyProbe returns a stub.dataForPath function that
+// answers /incident/comment and every /incident/feed request successfully
+// (the feed always carries want as an i_comm entry), sleeping delay on each
+// feed request and recording, via inFlight/maxInFlight, the peak number of
+// feed requests in flight at once. Tests use the recorded peak to prove
+// verifyIncidentCommentsWritten's requests actually overlap (and by how
+// much), which can't be observed from the command's exit code or output
+// alone.
+func incidentFeedConcurrencyProbe(want string, delay time.Duration, inFlight, maxInFlight *atomic.Int32) func(path string, body map[string]any) any {
+	return func(path string, body map[string]any) any {
+		switch path {
+		case "/incident/comment":
+			return map[string]any{}
+		case "/incident/feed":
+			cur := inFlight.Add(1)
+			defer inFlight.Add(-1)
+			for {
+				peak := maxInFlight.Load()
+				if cur <= peak || maxInFlight.CompareAndSwap(peak, cur) {
+					break
+				}
+			}
+			time.Sleep(delay)
+			return map[string]any{"items": []any{
+				map[string]any{"type": "i_comm", "created_at": 1, "detail": map[string]any{"comment": want}},
+			}}
+		default:
+			return map[string]any{}
+		}
+	}
+}
+
+// TestCommandIncidentCommentVerifyRunsConcurrently guards against a future
+// refactor silently reverting verifyIncidentCommentsWritten to a sequential
+// walk: with 4 incidents each sleeping on their /incident/feed request, a
+// sequential implementation would never have more than one request in flight
+// at once, while a correctly concurrent one will. Nothing about the command's
+// exit code or output would catch that regression — only observing overlap
+// directly, via incidentFeedConcurrencyProbe, does.
+func TestCommandIncidentCommentVerifyRunsConcurrently(t *testing.T) {
+	saveAndResetGlobals(t)
+	stub := newGFStub(t)
+	var inFlight, maxInFlight atomic.Int32
+	stub.dataForPath = incidentFeedConcurrencyProbe("the real comment", 20*time.Millisecond, &inFlight, &maxInFlight)
+
+	const n = 4
+	ids := make([]string, n)
+	for i := range ids {
+		ids[i] = fmt.Sprintf("inc-%d", i+1)
+	}
+	commentFile := writeCommentFile(t, "the real comment")
+
+	args := append([]string{"incident", "comment"}, ids...)
+	args = append(args, "--comment-file", commentFile)
+	out, err := execCommand(args...)
+	if err != nil {
+		t.Fatalf("[verify-concurrent] unexpected error: %v", err)
+	}
+	if !strings.Contains(out, fmt.Sprintf("Commented on %d incident(s).", n)) {
+		t.Fatalf("[verify-concurrent] unexpected output:\n%s", out)
+	}
+	if got := maxInFlight.Load(); got < 2 {
+		t.Fatalf("[verify-concurrent] observed peak in-flight verify requests = %d, want > 1 (verification never actually overlapped)", got)
+	}
+}
+
+// TestCommandIncidentCommentVerifyRespectsConcurrencyBound guards the other
+// side of the same fan-out: with 3x maxIncidentVerifyConcurrency incidents,
+// each holding its /incident/feed request open for a while, the peak observed
+// in flight must never exceed the worker limit — a regression that dropped
+// the semaphore (e.g. an unbounded "go func" per incident) would fire every
+// request at once and blow past it.
+func TestCommandIncidentCommentVerifyRespectsConcurrencyBound(t *testing.T) {
+	saveAndResetGlobals(t)
+	stub := newGFStub(t)
+	var inFlight, maxInFlight atomic.Int32
+	stub.dataForPath = incidentFeedConcurrencyProbe("the real comment", 20*time.Millisecond, &inFlight, &maxInFlight)
+
+	const n = maxIncidentVerifyConcurrency * 3
+	ids := make([]string, n)
+	for i := range ids {
+		ids[i] = fmt.Sprintf("inc-%d", i+1)
+	}
+	commentFile := writeCommentFile(t, "the real comment")
+
+	args := append([]string{"incident", "comment"}, ids...)
+	args = append(args, "--comment-file", commentFile)
+	out, err := execCommand(args...)
+	if err != nil {
+		t.Fatalf("[verify-bound] unexpected error: %v", err)
+	}
+	if !strings.Contains(out, fmt.Sprintf("Commented on %d incident(s).", n)) {
+		t.Fatalf("[verify-bound] unexpected output:\n%s", out)
+	}
+	if got := maxInFlight.Load(); got > maxIncidentVerifyConcurrency {
+		t.Fatalf("[verify-bound] observed peak in-flight verify requests = %d, want <= %d (the worker limit)", got, maxIncidentVerifyConcurrency)
+	}
+}
+
 // TestCommandIncidentLifecycleRejectsMoreThan100IDs covers the curated
 // commands that still enforce the 100-id batch cap client-side. unack and wake
 // were dropped in favor of their generated twins, which carry no client-side

internal/cli/gfstub_test.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"sync"
 	"testing"
 
 	"github.com/flashcatcloud/go-flashduty"
@@ -19,6 +20,16 @@ import (
 type gfStub struct {
 	server *httptest.Server
 
+	// mu guards every field below. Verified commands (e.g. incident comment's
+	// write-back check) now issue concurrent requests against a single stub
+	// server, so the handler below runs on more than one goroutine at once;
+	// without this lock the plain field writes here would race. It is
+	// deliberately released before invoking dataFor/dataForPath/data (see the
+	// handler below) so those calls run concurrently rather than being
+	// serialized by this lock — a test whose own closure touches shared state
+	// is responsible for synchronizing that state itself.
+	mu sync.Mutex
+
 	// lastPath is the path of the most recent request (no query string).
 	lastPath string
 	// lastBody is the decoded JSON body of the most recent request.
@@ -54,26 +65,39 @@ func newGFStub(t *testing.T) *gfStub {
 	t.Helper()
 	s := &gfStub{}
 	s.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var body map[string]any
+		if raw, err := io.ReadAll(r.Body); err == nil && len(raw) > 0 {
+			_ = json.Unmarshal(raw, &body)
+		}
+		path := r.URL.Path
+
+		s.mu.Lock()
 		s.requests++
-		s.lastPath = r.URL.Path
+		s.lastPath = path
 		s.lastAuthorization = r.Header.Get("Authorization")
-		s.lastBody = nil
-		if body, err := io.ReadAll(r.Body); err == nil && len(body) > 0 {
-			_ = json.Unmarshal(body, &s.lastBody)
-		}
-		s.bodies = append(s.bodies, s.lastBody)
+		s.lastBody = body
+		s.bodies = append(s.bodies, body)
+		dataForPath, dataFor, data := s.dataForPath, s.dataFor, s.data
+		s.mu.Unlock()
 
+		// dataForPath/dataFor/data run outside the lock: some tests (e.g. ones
+		// exercising verifyIncidentCommentsWritten's concurrent fan-out) rely on
+		// being able to observe genuine overlap between in-flight requests, which
+		// holding s.mu across the call would silently serialize away. A test
+		// closure that itself touches shared state is responsible for its own
+		// synchronization, same as any other concurrently-invoked callback.
 		var payload any
 		switch {
-		case s.dataForPath != nil:
-			payload = s.dataForPath(s.lastPath, s.lastBody)
-		case s.dataFor != nil:
-			payload = s.dataFor(s.lastBody)
-		case s.data != nil:
-			payload = s.data
+		case dataForPath != nil:
+			payload = dataForPath(path, body)
+		case dataFor != nil:
+			payload = dataFor(body)
+		case data != nil:
+			payload = data
 		default:
 			payload = map[string]any{}
 		}
+
 		resp := map[string]any{
 			"request_id": "test-request-id",
 			"error":      map[string]any{"code": "OK", "message": ""},

internal/cli/incident.go

@@ -9,6 +9,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/flashcatcloud/go-flashduty"
@@ -946,6 +947,13 @@ func resolveCommentFile(path string) (string, error) {
 // headroom for any real incident, and only bounds a pathological one.
 const maxIncidentFeedVerifyPages = 20
 
+// maxIncidentVerifyConcurrency bounds how many incidents
+// verifyIncidentCommentsWritten checks at once. These are network round-trips
+// against one API, not CPU work, so the bound is a fixed small number rather
+// than something scaled off NumCPU or exposed as a flag: it exists only to
+// avoid firing all of a 100-incident batch's requests at the same instant.
+const maxIncidentVerifyConcurrency = 8
+
 // commentVerificationGuidance is appended, verbatim, to every comment
 // verification failure — whether the entry simply couldn't be located (page
 // budget) or the re-fetch itself errored (transport failure). Both are
@@ -987,7 +995,18 @@ const commentVerificationNotFoundDetailFmt = "no timeline entry matches the writ
 // id before it can say what it says about the batch as a whole — reporting
 // only the first failure would leave any incident after it silently,
 // permanently unexamined while the agent believes it handled everything the
-// error mentioned.
+// error mentioned. This rules out an errgroup-style fail-fast fan-out: this
+// function's whole point is to keep going after a failure, not stop on one.
+//
+// Per-incident checks run concurrently, up to maxIncidentVerifyConcurrency at
+// a time, instead of walking ctx.Args one at a time: with a batch of up to
+// 100 incidents and up to maxIncidentFeedVerifyPages page fetches apiece, a
+// sequential walk could take thousands of round-trips before the command
+// returned. Each goroutine writes only its own slot of a pre-sized results
+// slice (index == position in ctx.Args), so the problems reported below
+// always come out in ctx.Args order regardless of which request happens to
+// finish first — the message must read the same on every run, not depend on
+// network timing.
 //
 // Neither a page-budget miss nor a transport error while re-fetching the feed
 // means the write was corrupted — corruption can no longer even be observed
@@ -1002,14 +1021,32 @@ const commentVerificationNotFoundDetailFmt = "no timeline entry matches the writ
 // simply couldn't be located within budget). The safe next step is to look,
 // not to write.
 func verifyIncidentCommentsWritten(ctx *RunContext, want string) error {
-	var problems []string
-	for _, id := range ctx.Args {
-		found, err := incidentTimelineHasComment(ctx, id, want)
-		switch {
-		case err != nil:
-			problems = append(problems, fmt.Sprintf("incident %s: %v", id, err))
-		case !found:
-			problems = append(problems, fmt.Sprintf("incident %s: "+commentVerificationNotFoundDetailFmt, id, maxIncidentFeedVerifyPages))
+	slots := make([]string, len(ctx.Args))
+
+	var wg sync.WaitGroup
+	sem := make(chan struct{}, maxIncidentVerifyConcurrency)
+	for i, id := range ctx.Args {
+		wg.Add(1)
+		go func(i int, id string) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+
+			found, err := incidentTimelineHasComment(ctx, id, want)
+			switch {
+			case err != nil:
+				slots[i] = fmt.Sprintf("incident %s: %v", id, err)
+			case !found:
+				slots[i] = fmt.Sprintf("incident %s: "+commentVerificationNotFoundDetailFmt, id, maxIncidentFeedVerifyPages)
+			}
+		}(i, id)
+	}
+	wg.Wait()
+
+	problems := make([]string, 0, len(slots))
+	for _, p := range slots {
+		if p != "" {
+			problems = append(problems, p)
 		}
 	}
 	if len(problems) == 0 {