fix

Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com>

Author: AdilFayyaz

runs/repository/impl/action.go

@@ -34,7 +34,6 @@ type actionRepo struct {
 	// Subscriber management for LISTEN/NOTIFY
 	runSubscribers    map[chan string]bool
 	actionSubscribers map[chan string]bool
-	abortSubscribers  map[chan string]bool
 	mu                sync.RWMutex
 }
 
@@ -50,7 +49,6 @@ func NewActionRepo(db *gorm.DB, dbConfig database.DbConfig) interfaces.ActionRep
 		pgConfig:          dbConfig.Postgres,
 		runSubscribers:    make(map[chan string]bool),
 		actionSubscribers: make(map[chan string]bool),
-		abortSubscribers:  make(map[chan string]bool),
 	}
 
 	// Start LISTEN/NOTIFY for PostgreSQL
@@ -274,9 +272,8 @@ func (r *actionRepo) AbortRun(ctx context.Context, runID *common.RunIdentifier,
 		return fmt.Errorf("failed to abort run: %w", result.Error)
 	}
 
-	// Notify run subscribers and abort reconciler.
+	// Notify run subscribers.
 	r.notifyRunUpdate(ctx, runID)
-	r.notifyAbortRequest(ctx, &common.ActionIdentifier{Run: runID, Name: runID.Name})
 
 	logger.Infof(ctx, "Aborted run: %s/%s/%s/%s", runID.Org, runID.Project, runID.Domain, runID.Name)
 	return nil
@@ -476,17 +473,16 @@ func (r *actionRepo) AbortAction(ctx context.Context, actionID *common.ActionIde
 
 	result := r.db.WithContext(ctx).
 		Model(&models.Action{}).
-		Where("org = ? AND project = ? AND domain = ? AND name = ?",
-			actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Name).
+		Where("org = ? AND project = ? AND domain = ? AND run_name = ? AND name = ?",
+			actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Run.Name, actionID.Name).
 		Updates(updates)
 
 	if result.Error != nil {
 		return fmt.Errorf("failed to abort action: %w", result.Error)
 	}
 
-	// Notify action subscribers and abort reconciler.
+	// Notify action subscribers.
 	r.notifyActionUpdate(ctx, actionID)
-	r.notifyAbortRequest(ctx, actionID)
 
 	logger.Infof(ctx, "Aborted action: %s", actionID.Name)
 	return nil
@@ -509,8 +505,8 @@ func (r *actionRepo) ListPendingAborts(ctx context.Context) ([]*models.Action, e
 func (r *actionRepo) MarkAbortAttempt(ctx context.Context, actionID *common.ActionIdentifier) (int, error) {
 	result := r.db.WithContext(ctx).
 		Model(&models.Action{}).
-		Where("org = ? AND project = ? AND domain = ? AND name = ?",
-			actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Name).
+		Where("org = ? AND project = ? AND domain = ? AND run_name = ? AND name = ?",
+			actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Run.Name, actionID.Name).
 		Updates(map[string]interface{}{
 			"abort_attempt_count": gorm.Expr("abort_attempt_count + 1"),
 			"updated_at":          time.Now(),
@@ -523,8 +519,8 @@ func (r *actionRepo) MarkAbortAttempt(ctx context.Context, actionID *common.Acti
 	var action models.Action
 	if err := r.db.WithContext(ctx).
 		Select("abort_attempt_count").
-		Where("org = ? AND project = ? AND domain = ? AND name = ?",
-			actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Name).
+		Where("org = ? AND project = ? AND domain = ? AND run_name = ? AND name = ?",
+			actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Run.Name, actionID.Name).
 		First(&action).Error; err != nil {
 		return 0, fmt.Errorf("failed to read abort attempt count: %w", err)
 	}
@@ -535,8 +531,8 @@ func (r *actionRepo) MarkAbortAttempt(ctx context.Context, actionID *common.Acti
 func (r *actionRepo) ClearAbortRequest(ctx context.Context, actionID *common.ActionIdentifier) error {
 	result := r.db.WithContext(ctx).
 		Model(&models.Action{}).
-		Where("org = ? AND project = ? AND domain = ? AND name = ?",
-			actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Name).
+		Where("org = ? AND project = ? AND domain = ? AND run_name = ? AND name = ?",
+			actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Run.Name, actionID.Name).
 		Updates(map[string]interface{}{
 			"abort_requested_at":  nil,
 			"abort_attempt_count": 0,
@@ -549,78 +545,6 @@ func (r *actionRepo) ClearAbortRequest(ctx context.Context, actionID *common.Act
 	return nil
 }
 
-// WatchAbortRequests delivers abort_requests NOTIFY payloads to the caller.
-// On non-Postgres setups it polls for pending aborts every 5 seconds.
-func (r *actionRepo) WatchAbortRequests(ctx context.Context, payloads chan<- string, errs chan<- error) {
-	if r.isPostgres {
-		notifCh := make(chan string, 20)
-
-		r.mu.Lock()
-		r.abortSubscribers[notifCh] = true
-		r.mu.Unlock()
-
-		defer func() {
-			r.mu.Lock()
-			delete(r.abortSubscribers, notifCh)
-			close(notifCh)
-			r.mu.Unlock()
-		}()
-
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case payload := <-notifCh:
-				select {
-				case payloads <- payload:
-				case <-ctx.Done():
-					return
-				}
-			}
-		}
-	} else {
-		// SQLite fallback: poll for pending aborts periodically.
-		ticker := time.NewTicker(5 * time.Second)
-		defer ticker.Stop()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-ticker.C:
-				actions, err := r.ListPendingAborts(ctx)
-				if err != nil {
-					select {
-					case errs <- err:
-					default:
-					}
-					continue
-				}
-				for _, a := range actions {
-					payload := fmt.Sprintf("%s/%s/%s/%s/%s", a.Org, a.Project, a.Domain, a.RunName, a.Name)
-					select {
-					case payloads <- payload:
-					case <-ctx.Done():
-						return
-					}
-				}
-			}
-		}
-	}
-}
-
-// notifyAbortRequest fires NOTIFY abort_requests so the AbortReconciler wakes immediately.
-func (r *actionRepo) notifyAbortRequest(ctx context.Context, actionID *common.ActionIdentifier) {
-	if !r.isPostgres {
-		return
-	}
-	payload := fmt.Sprintf("%s/%s/%s/%s/%s",
-		actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Run.Name, actionID.Name)
-	notifySQL := fmt.Sprintf("NOTIFY abort_requests, '%s'", payload)
-	if err := r.db.WithContext(ctx).Exec(notifySQL).Error; err != nil {
-		logger.Errorf(ctx, "Failed to NOTIFY abort_requests: %v", err)
-	}
-}
-
 // UpdateActionState updates the state of an action
 func (r *actionRepo) UpdateActionState(ctx context.Context, actionID *common.ActionIdentifier, state string) error {
 	// Parse the state JSON to extract the phase
@@ -953,11 +877,6 @@ func (r *actionRepo) startPostgresListener() {
 		return
 	}
 
-	if err := r.listener.Listen("abort_requests"); err != nil {
-		logger.Errorf(context.Background(), "Failed to listen to abort_requests: %v", err)
-		return
-	}
-
 	logger.Infof(context.Background(), "PostgreSQL LISTEN/NOTIFY started")
 
 	// Process notifications
@@ -994,17 +913,6 @@ func (r *actionRepo) startPostgresListener() {
 				}
 				r.mu.RUnlock()
 
-			case "abort_requests":
-				// Broadcast to all abort reconciler subscribers
-				r.mu.RLock()
-				for ch := range r.abortSubscribers {
-					select {
-					case ch <- notif.Extra:
-					default:
-						logger.Warnf(context.Background(), "Abort subscriber channel full, dropping notification")
-					}
-				}
-				r.mu.RUnlock()
 			}
 
 		case <-time.After(90 * time.Second):

runs/repository/interfaces/action.go

@@ -30,7 +30,6 @@ type ActionRepo interface {
 	ListPendingAborts(ctx context.Context) ([]*models.Action, error)
 	MarkAbortAttempt(ctx context.Context, actionID *common.ActionIdentifier) (attemptCount int, err error)
 	ClearAbortRequest(ctx context.Context, actionID *common.ActionIdentifier) error
-	WatchAbortRequests(ctx context.Context, payloads chan<- string, errs chan<- error)
 
 	// Watch operations (for streaming)
 	WatchRunUpdates(ctx context.Context, runID *common.RunIdentifier, updates chan<- *models.Run, errs chan<- error)

runs/repository/mocks/action_repo.go

@@ -3,7 +3,6 @@ package service
 import (
 	"context"
 	"fmt"
-	"strings"
 	"sync"
 	"time"
 
@@ -85,8 +84,8 @@ func (q *dedupeQueue) push(ctx context.Context, task abortTask) bool {
 }
 
 // scheduleRequeue re-enqueues the task after delay.
-// The key remains in the set during the wait window so that any NOTIFY arriving during
-// the backoff is correctly deduped (no duplicate processing).
+// The key remains in the set during the wait window so that any Push call arriving
+// during the backoff is correctly deduped (no duplicate processing).
 func (q *dedupeQueue) scheduleRequeue(ctx context.Context, task abortTask, delay time.Duration) {
 	time.AfterFunc(delay, func() {
 		// Remove then re-push so push's dedup check passes.
@@ -139,6 +138,14 @@ func NewAbortReconciler(repo interfaces.Repository, actionsClient actionsconnect
 	}
 }
 
+// Push enqueues an abort request for the given action. Safe to call concurrently.
+// No-op if the key is already queued (dedup).
+func (r *AbortReconciler) Push(ctx context.Context, actionID *common.ActionIdentifier, reason string) {
+	key := fmt.Sprintf("%s/%s/%s/%s/%s",
+		actionID.Run.Org, actionID.Run.Project, actionID.Run.Domain, actionID.Run.Name, actionID.Name)
+	r.queue.push(ctx, abortTask{actionID: actionID, reason: reason, key: key})
+}
+
 // Run starts the reconciler. It blocks until ctx is cancelled.
 func (r *AbortReconciler) Run(ctx context.Context) error {
 	logger.Infof(ctx, "AbortReconciler starting (%d workers, max %d attempts)", r.cfg.Workers, r.cfg.MaxAttempts)
@@ -155,46 +162,14 @@ func (r *AbortReconciler) Run(ctx context.Context) error {
 		}()
 	}
 
-	// Startup scan: enqueue any actions that were left pending from before this process started.
+	// Startup scan: enqueue any actions left pending before this process started (crash recovery).
 	if err := r.startupScan(ctx); err != nil {
 		logger.Errorf(ctx, "AbortReconciler startup scan failed: %v", err)
-		// Non-fatal — the NOTIFY watcher will still pick up new aborts.
 	}
 
-	// Watch for new abort requests via NOTIFY (or polling on SQLite).
-	payloads := make(chan string, 50)
-	errs := make(chan error, 10)
-	go r.repo.ActionRepo().WatchAbortRequests(ctx, payloads, errs)
-
-	for {
-		select {
-		case <-ctx.Done():
-			wg.Wait()
-			return nil
-		case err := <-errs:
-			logger.Warnf(ctx, "AbortReconciler watch error: %v", err)
-		case payload := <-payloads:
-			task, err := parseAbortPayload(payload)
-			if err != nil {
-				logger.Warnf(ctx, "AbortReconciler: invalid payload %q: %v", payload, err)
-				continue
-			}
-			// Fetch current abort_reason from DB (NOTIFY payload does not carry reason).
-			action, err := r.repo.ActionRepo().GetAction(ctx, task.actionID)
-			if err != nil {
-				logger.Warnf(ctx, "AbortReconciler: failed to fetch action %s: %v", task.key, err)
-				continue
-			}
-			if action.AbortRequestedAt == nil {
-				// Already cleared (race between scan and notify) — skip.
-				continue
-			}
-			if action.AbortReason != nil {
-				task.reason = *action.AbortReason
-			}
-			r.queue.push(ctx, task)
-		}
-	}
+	<-ctx.Done()
+	wg.Wait()
+	return nil
 }
 
 // startupScan enqueues all actions that have abort_requested_at set.
@@ -297,23 +272,3 @@ func isAlreadyTerminated(err error) bool {
 	}
 	return connectErr.Code() == connect.CodeNotFound
 }
-
-// parseAbortPayload parses "org/project/domain/run/actionName" into an abortTask.
-func parseAbortPayload(payload string) (abortTask, error) {
-	parts := strings.SplitN(payload, "/", 5)
-	if len(parts) != 5 {
-		return abortTask{}, fmt.Errorf("expected 5 parts, got %d", len(parts))
-	}
-	return abortTask{
-		key: payload,
-		actionID: &common.ActionIdentifier{
-			Run: &common.RunIdentifier{
-				Org:     parts[0],
-				Project: parts[1],
-				Domain:  parts[2],
-				Name:    parts[3],
-			},
-			Name: parts[4],
-		},
-	}, nil
-}

runs/service/abort_reconciler.go

@@ -18,21 +18,12 @@ import (
 )
 
 // newTestReconciler builds a reconciler wired to mocks with fast timing for tests.
-// WatchAbortRequests is set up to block until ctx is cancelled so tests control
-// the queue via the startup scan and direct queue pushes.
 func newTestReconciler(t *testing.T) (*repoMocks.ActionRepo, *mockActionsClient, *AbortReconciler) {
 	t.Helper()
 	actionRepo := repoMocks.NewActionRepo(t)
 	repo := repoMocks.NewRepository(t)
 	repo.On("ActionRepo").Return(actionRepo).Maybe()
 
-	actionRepo.On("WatchAbortRequests", mock.Anything, mock.Anything, mock.Anything).
-		Run(func(args mock.Arguments) {
-			ctx := args.Get(0).(context.Context)
-			<-ctx.Done()
-		}).
-		Return().Maybe()
-
 	actionsClient := &mockActionsClient{}
 
 	reconciler := NewAbortReconciler(repo, actionsClient, AbortReconcilerConfig{
@@ -168,11 +159,7 @@ func TestAbortReconciler_DeduplicatesQueue(t *testing.T) {
 
 	// Push a duplicate while the first is being processed.
 	time.Sleep(5 * time.Millisecond)
-	reconciler.queue.push(ctx, abortTask{
-		actionID: actionID,
-		key:      "org/proj/dev/rtest1/rtest1",
-		reason:   "dup",
-	})
+	reconciler.Push(ctx, actionID, "dup")
 
 	assert.Eventually(t, cleared.Load, 400*time.Millisecond, 5*time.Millisecond)
 	// Abort should only have been called once.