fix(task): make review finding status updates idempotent on resolved_at (#3670)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-15 17:31:55 +08:00
committed by GitHub
parent b2efea3892
commit ad34d0d1e0
10 changed files with 427 additions and 34 deletions
@@ -298,19 +298,32 @@ func (r *Repository) GetTaskReviewFinding(ctx context.Context, findingID string)
return f, err
}
// UpdateTaskReviewFindingStatus sets a finding's status and resolved_at.
func (r *Repository) UpdateTaskReviewFindingStatus(ctx context.Context, findingID string, status models.ReviewFindingStatus, resolvedAt *time.Time) error {
result, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE task_review_findings SET status = ?, resolved_at = ?, updated_at = ? WHERE id = ?
`), string(status), resolvedAt, time.Now().UTC(), findingID)
// TransitionTaskReviewFindingStatus updates a finding and returns the stored
// row. The CASE expression reads the old status inside the same database write,
// so concurrent callers cannot apply a timestamp decision from stale state.
func (r *Repository) TransitionTaskReviewFindingStatus(ctx context.Context, findingID string, status models.ReviewFindingStatus) (*models.TaskReviewFinding, error) {
now := time.Now().UTC()
row := r.db.QueryRowxContext(ctx, r.db.Rebind(`
UPDATE task_review_findings
SET status = ?,
resolved_at = CASE
WHEN status = ? THEN resolved_at
WHEN ? IN ('resolved', 'dismissed') THEN ?
ELSE NULL
END,
updated_at = ?
WHERE id = ?
RETURNING `+reviewFindingColumns),
string(status), string(status), string(status), now,
now, findingID)
finding, err := scanReviewFinding(row)
if errors.Is(err, sql.ErrNoRows) {
return nil, fmt.Errorf("%w: %s", models.ErrTaskReviewFindingNotFound, findingID)
}
if err != nil {
return fmt.Errorf("failed to update task review finding status: %w", err)
return nil, fmt.Errorf("failed to transition task review finding status: %w", err)
}
rows, _ := result.RowsAffected()
if rows == 0 {
return fmt.Errorf("%w: %s", models.ErrTaskReviewFindingNotFound, findingID)
}
return nil
return finding, nil
}
// DeleteSupersededTaskReviewFindings removes still-open findings from earlier
@@ -0,0 +1,76 @@
package sqlite
import (
"context"
"sync"
"testing"
"github.com/kandev/kandev/internal/task/models"
)
// TestTaskReviewFinding_ConcurrentTransitionsKeepResolvedTimestamp proves that
// concurrent duplicate resolutions return the same persisted timestamp and do
// not overwrite each other's resolved_at value.
func TestTaskReviewFinding_ConcurrentTransitionsKeepResolvedTimestamp(t *testing.T) {
repo := newRepoForSessionTests(t)
ctx := context.Background()
seedReviewTask(t, ctx, repo, "task-status-concurrent")
run := newReviewRun(t, ctx, repo, "task-status-concurrent")
f := finding(run.ID, "task-status-concurrent", "a.go", "issue", 4)
if err := repo.CreateTaskReviewFindings(ctx, []*models.TaskReviewFinding{f}); err != nil {
t.Fatalf("CreateTaskReviewFindings: %v", err)
}
start := make(chan struct{})
results := make(chan struct {
finding *models.TaskReviewFinding
err error
}, 2)
var wg sync.WaitGroup
for range 2 {
wg.Add(1)
go func() {
defer wg.Done()
<-start
got, err := repo.TransitionTaskReviewFindingStatus(ctx, f.ID, models.ReviewFindingResolved)
results <- struct {
finding *models.TaskReviewFinding
err error
}{finding: got, err: err}
}()
}
close(start)
wg.Wait()
close(results)
var resolvedAt *models.TaskReviewFinding
for result := range results {
if result.err != nil {
t.Fatalf("concurrent transition: %v", result.err)
}
if result.finding == nil || result.finding.Status != models.ReviewFindingResolved || result.finding.ResolvedAt == nil {
t.Fatalf("concurrent transition returned invalid finding: %+v", result.finding)
}
if resolvedAt == nil {
resolvedAt = result.finding
continue
}
if !result.finding.ResolvedAt.Equal(*resolvedAt.ResolvedAt) {
t.Fatalf("concurrent transitions returned different resolved_at values: %v and %v", resolvedAt.ResolvedAt, result.finding.ResolvedAt)
}
}
final, err := repo.GetTaskReviewFinding(ctx, f.ID)
if err != nil {
t.Fatalf("GetTaskReviewFinding: %v", err)
}
if final.Status != models.ReviewFindingResolved || final.ResolvedAt == nil {
t.Fatalf("final finding is not resolved: %+v", final)
}
if resolvedAt == nil {
t.Fatal("concurrent transitions returned no findings")
}
if !final.ResolvedAt.Equal(*resolvedAt.ResolvedAt) {
t.Fatalf("final resolved_at=%v, want %v", final.ResolvedAt, resolvedAt.ResolvedAt)
}
}
@@ -0,0 +1,81 @@
package sqlite
import (
"context"
"sync"
"testing"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/testutil"
)
// TestPostgresTaskReviewFinding_ConcurrentTransitionsKeepResolvedTimestamp
// exercises the same transition through two real PostgreSQL connections.
// It skips unless KANDEV_TEST_POSTGRES_DSN is set.
func TestPostgresTaskReviewFinding_ConcurrentTransitionsKeepResolvedTimestamp(t *testing.T) {
db := openIsolatedPostgresMultiConn(t, testutil.PostgresDSNFromEnv(t), 2)
repo, err := NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("init postgres schema: %v", err)
}
ctx := context.Background()
seedReviewTask(t, ctx, repo, "task-status-postgres-concurrent")
run := newReviewRun(t, ctx, repo, "task-status-postgres-concurrent")
f := finding(run.ID, "task-status-postgres-concurrent", "a.go", "issue", 4)
if err := repo.CreateTaskReviewFindings(ctx, []*models.TaskReviewFinding{f}); err != nil {
t.Fatalf("CreateTaskReviewFindings: %v", err)
}
start := make(chan struct{})
results := make(chan struct {
finding *models.TaskReviewFinding
err error
}, 2)
var wg sync.WaitGroup
for range 2 {
wg.Add(1)
go func() {
defer wg.Done()
<-start
got, err := repo.TransitionTaskReviewFindingStatus(ctx, f.ID, models.ReviewFindingResolved)
results <- struct {
finding *models.TaskReviewFinding
err error
}{finding: got, err: err}
}()
}
close(start)
wg.Wait()
close(results)
var resolvedAt *models.TaskReviewFinding
for result := range results {
if result.err != nil {
t.Fatalf("concurrent transition: %v", result.err)
}
if result.finding == nil || result.finding.Status != models.ReviewFindingResolved || result.finding.ResolvedAt == nil {
t.Fatalf("concurrent transition returned invalid finding: %+v", result.finding)
}
if resolvedAt == nil {
resolvedAt = result.finding
continue
}
if !result.finding.ResolvedAt.Equal(*resolvedAt.ResolvedAt) {
t.Fatalf("concurrent transitions returned different resolved_at values: %v and %v", resolvedAt.ResolvedAt, result.finding.ResolvedAt)
}
}
final, err := repo.GetTaskReviewFinding(ctx, f.ID)
if err != nil {
t.Fatalf("GetTaskReviewFinding: %v", err)
}
if final.Status != models.ReviewFindingResolved || final.ResolvedAt == nil {
t.Fatalf("final finding is not resolved: %+v", final)
}
if resolvedAt == nil {
t.Fatal("concurrent transitions returned no findings")
}
if !final.ResolvedAt.Equal(*resolvedAt.ResolvedAt) {
t.Fatalf("final resolved_at=%v, want %v", final.ResolvedAt, resolvedAt.ResolvedAt)
}
}
@@ -199,30 +199,23 @@ func TestTaskReviewFinding_UpdateStatus(t *testing.T) {
t.Fatalf("CreateTaskReviewFindings: %v", err)
}
now := f.CreatedAt
if err := repo.UpdateTaskReviewFindingStatus(ctx, f.ID, models.ReviewFindingResolved, &now); err != nil {
t.Fatalf("UpdateTaskReviewFindingStatus: %v", err)
}
got, err := repo.GetTaskReviewFinding(ctx, f.ID)
got, err := repo.TransitionTaskReviewFindingStatus(ctx, f.ID, models.ReviewFindingResolved)
if err != nil {
t.Fatalf("GetTaskReviewFinding: %v", err)
t.Fatalf("TransitionTaskReviewFindingStatus: %v", err)
}
if got.Status != models.ReviewFindingResolved || got.ResolvedAt == nil {
t.Fatalf("expected resolved with resolved_at, got %+v", got)
}
if err := repo.UpdateTaskReviewFindingStatus(ctx, f.ID, models.ReviewFindingOpen, nil); err != nil {
t.Fatalf("reopen: %v", err)
}
got, err = repo.GetTaskReviewFinding(ctx, f.ID)
got, err = repo.TransitionTaskReviewFindingStatus(ctx, f.ID, models.ReviewFindingOpen)
if err != nil {
t.Fatalf("GetTaskReviewFinding after reopen: %v", err)
t.Fatalf("reopen: %v", err)
}
if got.Status != models.ReviewFindingOpen || got.ResolvedAt != nil {
t.Fatalf("expected reopen to clear resolved_at, got %+v", got)
}
err = repo.UpdateTaskReviewFindingStatus(ctx, "missing", models.ReviewFindingOpen, nil)
_, err = repo.TransitionTaskReviewFindingStatus(ctx, "missing", models.ReviewFindingOpen)
if !errors.Is(err, models.ErrTaskReviewFindingNotFound) {
t.Fatalf("expected ErrTaskReviewFindingNotFound, got %v", err)
}
@@ -50,7 +50,7 @@ type reviewRepo interface {
CreateTaskReviewFindings(ctx context.Context, findings []*models.TaskReviewFinding) error
ListTaskReviewFindings(ctx context.Context, taskID string) ([]*models.TaskReviewFinding, error)
GetTaskReviewFinding(ctx context.Context, findingID string) (*models.TaskReviewFinding, error)
UpdateTaskReviewFindingStatus(ctx context.Context, findingID string, status models.ReviewFindingStatus, resolvedAt *time.Time) error
TransitionTaskReviewFindingStatus(ctx context.Context, findingID string, status models.ReviewFindingStatus) (*models.TaskReviewFinding, error)
DeleteSupersededTaskReviewFindings(ctx context.Context, taskID, runID string, keys []models.ReviewFindingKey) ([]string, error)
DeleteTaskReviewByTask(ctx context.Context, taskID string) error
}
@@ -468,7 +468,9 @@ func buildReviewFinding(taskID, runID string, in ReviewFindingInput) (*models.Ta
}
// UpdateFindingStatus records the human's disposition of a finding. Moving to
// resolved stamps resolved_at; returning to open clears it.
// resolved or dismissed stamps resolved_at; returning to open clears it.
// Resubmitting the finding's current status is idempotent and leaves
// resolved_at unchanged rather than re-stamping it to now.
func (s *ReviewService) UpdateFindingStatus(ctx context.Context, findingID string, status models.ReviewFindingStatus) (*models.TaskReviewFinding, error) {
if findingID == "" {
return nil, fmt.Errorf("%w: finding id is required", ErrReviewFindingNotFound)
@@ -476,15 +478,7 @@ func (s *ReviewService) UpdateFindingStatus(ctx context.Context, findingID strin
if !models.ValidReviewFindingStatus(status) {
return nil, fmt.Errorf("%w: unknown status %q", ErrInvalidReviewFinding, status)
}
var resolvedAt *time.Time
if status == models.ReviewFindingResolved || status == models.ReviewFindingDismissed {
now := time.Now().UTC()
resolvedAt = &now
}
if err := s.repo.UpdateTaskReviewFindingStatus(ctx, findingID, status, resolvedAt); err != nil {
return nil, err
}
finding, err := s.repo.GetTaskReviewFinding(ctx, findingID)
finding, err := s.repo.TransitionTaskReviewFindingStatus(ctx, findingID, status)
if err != nil {
return nil, err
}
@@ -0,0 +1,116 @@
package service
import (
"context"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/kandev/kandev/internal/task/models"
)
// orderedFindingRepo forces two status requests to reach the repository before
// it applies the reopen-then-resolve ordering that exposes stale state.
type orderedFindingRepo struct {
reviewRepo
transitionsReady chan struct{}
releaseTransitions chan struct{}
transitionCalls int32
openApplied chan struct{}
openOnce sync.Once
}
func (r *orderedFindingRepo) awaitTransitionStart() {
n := atomic.AddInt32(&r.transitionCalls, 1)
if n > 2 {
return
}
if n == 2 {
close(r.transitionsReady)
}
<-r.releaseTransitions
}
func (r *orderedFindingRepo) TransitionTaskReviewFindingStatus(
ctx context.Context,
findingID string,
status models.ReviewFindingStatus,
) (*models.TaskReviewFinding, error) {
r.awaitTransitionStart()
transitioner, ok := r.reviewRepo.(interface {
TransitionTaskReviewFindingStatus(context.Context, string, models.ReviewFindingStatus) (*models.TaskReviewFinding, error)
})
if !ok {
return nil, nil
}
if status == models.ReviewFindingResolved {
<-r.openApplied
}
finding, err := transitioner.TransitionTaskReviewFindingStatus(ctx, findingID, status)
if status == models.ReviewFindingOpen {
r.openOnce.Do(func() { close(r.openApplied) })
}
return finding, err
}
func TestReviewService_UpdateFindingStatusDoesNotRestoreStaleResolvedAt(t *testing.T) {
base, eventBus, repo := createTestReviewService(t)
ctx := context.Background()
seedTask(t, ctx, repo, "task-status-race")
_, findings, err := base.PublishFindings(ctx, PublishFindingsRequest{
TaskID: "task-status-race",
Findings: []ReviewFindingInput{{
FilePath: "a.go", StartLine: 1, EndLine: 1, Severity: "major",
Category: "correctness", Title: "stale", Body: "stale state",
}},
})
if err != nil {
t.Fatalf("PublishFindings: %v", err)
}
initial, err := base.UpdateFindingStatus(ctx, findings[0].ID, models.ReviewFindingResolved)
if err != nil || initial.ResolvedAt == nil {
t.Fatalf("seed resolved finding: finding=%+v err=%v", initial, err)
}
initialResolvedAt := *initial.ResolvedAt
ordered := &orderedFindingRepo{
reviewRepo: repo,
transitionsReady: make(chan struct{}),
releaseTransitions: make(chan struct{}),
openApplied: make(chan struct{}),
}
svc := NewReviewService(ordered, eventBus, base.logger)
results := make(chan error, 2)
go func() {
_, updateErr := svc.UpdateFindingStatus(ctx, findings[0].ID, models.ReviewFindingOpen)
results <- updateErr
}()
go func() {
_, updateErr := svc.UpdateFindingStatus(ctx, findings[0].ID, models.ReviewFindingResolved)
results <- updateErr
}()
select {
case <-ordered.transitionsReady:
case <-time.After(5 * time.Second):
t.Fatal("status requests did not reach the transition barrier")
}
close(ordered.releaseTransitions)
for range 2 {
if updateErr := <-results; updateErr != nil {
t.Fatalf("UpdateFindingStatus: %v", updateErr)
}
}
final, err := repo.GetTaskReviewFinding(ctx, findings[0].ID)
if err != nil {
t.Fatalf("GetTaskReviewFinding: %v", err)
}
if final.Status != models.ReviewFindingResolved || final.ResolvedAt == nil {
t.Fatalf("expected the final status to be resolved with a timestamp, got %+v", final)
}
if final.ResolvedAt.Equal(initialResolvedAt) {
t.Fatalf("stale resolve restored the old timestamp %v", final.ResolvedAt)
}
}
@@ -5,6 +5,7 @@ import (
"errors"
"strings"
"testing"
"time"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/events"
@@ -443,6 +444,42 @@ func TestReviewService_UpdateFindingStatusStampsAndClearsResolvedAt(t *testing.T
}
}
func TestReviewService_UpdateFindingStatusResubmittingCurrentStatusIsIdempotent(t *testing.T) {
svc, eventBus, repo := createTestReviewService(t)
ctx := context.Background()
seedTask(t, ctx, repo, "task-12b")
_, findings, err := svc.PublishFindings(ctx, PublishFindingsRequest{
TaskID: "task-12b",
Findings: []ReviewFindingInput{validFindingInput()},
})
if err != nil {
t.Fatalf("PublishFindings: %v", err)
}
resolved, err := svc.UpdateFindingStatus(ctx, findings[0].ID, models.ReviewFindingResolved)
if err != nil {
t.Fatalf("resolve: %v", err)
}
firstResolvedAt := resolved.ResolvedAt
if firstResolvedAt == nil {
t.Fatal("expected resolved_at to be stamped")
}
time.Sleep(2 * time.Millisecond)
eventBus.ClearEvents()
resubmitted, err := svc.UpdateFindingStatus(ctx, findings[0].ID, models.ReviewFindingResolved)
if err != nil {
t.Fatalf("resubmit: %v", err)
}
if resubmitted.ResolvedAt == nil || !resubmitted.ResolvedAt.Equal(*firstResolvedAt) {
t.Fatalf("resubmitting the current status must leave resolved_at unchanged, got %v want %v", resubmitted.ResolvedAt, firstResolvedAt)
}
if countEvents(eventBus.GetPublishedEvents(), events.TaskReviewFindingUpdated) != 1 {
t.Fatalf("resubmitting the current status should still publish an update event, got %v", eventTypes(eventBus.GetPublishedEvents()))
}
}
func TestReviewService_UpdateFindingStatusRejectsUnknownStatus(t *testing.T) {
svc, _, repo := createTestReviewService(t)
ctx := context.Background()
@@ -0,0 +1,28 @@
---
created: 2026-09-14
status: completed
requirements:
- REQ-AGENTS-NATIVE-CODE-REVIEW-001
system_design:
- ../../specs/agents/system-design/native-code-review-status-atomicity.md
legacy_specs: []
---
# Implementation plan: Native review finding status atomicity
Review finding status changes must be safe when more than one client updates the
same finding at the same time. The repository owns one atomic transition that
reads the stored status, derives `resolved_at`, and returns the stored finding.
The service keeps validation and publishes the existing update event after the
transition succeeds.
## Work order
- [x] [Make review finding status transitions atomic](task-01-review-finding-status-atomicity.md)
## Scope
- Keep the existing review finding and WebSocket contracts.
- Use one repository transition for status and timestamp updates.
- Cover concurrent SQLite writers and an environment-gated PostgreSQL
multi-connection path.
@@ -0,0 +1,36 @@
---
id: review-finding-status-atomicity
title: Make review finding status transitions atomic
status: completed
wave: 1
depends_on: []
plan: plan.md
requirements:
- REQ-AGENTS-NATIVE-CODE-REVIEW-001
acceptance_criteria:
- AC-AGENTS-NATIVE-CODE-REVIEW-001.2
- AC-AGENTS-NATIVE-CODE-REVIEW-001.4
system_design:
- ../../specs/agents/system-design/native-code-review-status-atomicity.md
---
# Make review finding status transitions atomic
## Scope
Replace the review service read/update/read sequence with one repository
transition. The SQL update derives `resolved_at` from the status stored by that
same statement and returns the final finding row. Keep status validation and
`TaskReviewFindingUpdated` publication in the service.
## Verification
- Review service concurrent update coverage passes.
- SQLite repository concurrent transition coverage passes.
- The real PostgreSQL two-connection test runs when its environment flag is
enabled and remains skipped by default.
- The public WebSocket payload remains unchanged.
## Result
The implementation is complete at commit `674dd00f9dae51d1982870ae88583390e1e3aa68`.
@@ -0,0 +1,19 @@
---
status: current
system: agents
requirements:
- REQ-AGENTS-NATIVE-CODE-REVIEW-001
---
# Native review finding status atomicity
The task review repository is the owner of persisted finding state. A status
transition updates `status` and `resolved_at` in one database statement and
returns the row produced by that statement. The service validates the requested
status before it calls the repository, then publishes the existing finding
updated event with the returned row.
Concurrent writers can therefore observe different final rows without creating
a timestamp that describes an earlier read. SQLite tests provide deterministic
coverage. A PostgreSQL test with two database connections checks the same
transaction boundary when a PostgreSQL environment is available.