Skip to content

fix(runs): resync watch streams when notifications are lost #8240

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ChickenBenny wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(runs): resync watch streams when notifications are lost #8240

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
aff761d
refactor(runs): track watch subscribers as subscription structs
ChickenBenny Oct 24, 2026
a94be3e
fix(runs): resync watch subscribers after the listener reconnects
ChickenBenny Oct 24, 2026
3ccb366
fix(runs): resync list watchers after the listener reconnects
ChickenBenny Oct 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
192 changes: 105 additions & 87 deletions runs/repository/impl/action.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,8 +124,8 @@ type actionRepo struct {
listenerMinReconnectInterval time.Duration

// Subscriber management for LISTEN/NOTIFY
runSubscribers map[chan string]bool
actionSubscribers map[chan string]bool
runSubscribers map[*subscription]struct{}
actionSubscribers map[*subscription]struct{}
mu sync.RWMutex

notifyMu sync.Mutex
Expand Down Expand Up @@ -158,8 +158,8 @@ func NewActionRepo(
repo := &actionRepo{
db: db,
dsn: dsn,
runSubscribers: make(map[chan string]bool),
actionSubscribers: make(map[chan string]bool),
runSubscribers: make(map[*subscription]struct{}),
actionSubscribers: make(map[*subscription]struct{}),
notificationConfig: notificationConfig,
notificationMetrics: notificationMetrics,
}
Expand Down Expand Up @@ -780,66 +780,88 @@ func (r *actionRepo) WatchStateUpdates(ctx context.Context, updates chan<- *comm
<-ctx.Done()
}

// WatchRunUpdates watches for run updates via LISTEN/NOTIFY
func (r *actionRepo) WatchRunUpdates(ctx context.Context, runID *common.RunIdentifier, updates chan<- *models.Run, errs chan<- error) {
runKey := fmt.Sprintf("%s/%s/%s", runID.Project, runID.Domain, runID.Name)
notifCh := make(chan string, r.notificationConfig.watchBufferSize)
// subscription is one Watch* subscriber.
type subscription struct {
notify chan string
// resync asks the subscriber to re-read its state from the DB after the listener reconnects.
resync chan struct{}
}

// requestResync never blocks, and repeated requests coalesce into one re-read.
func (s *subscription) requestResync() {
select {
case s.resync <- struct{}{}:
default:
}
}

func (r *actionRepo) subscribe(subs map[*subscription]struct{}) *subscription {
sub := &subscription{
notify: make(chan string, r.notificationConfig.watchBufferSize),
resync: make(chan struct{}, 1),
}
r.mu.Lock()
r.runSubscribers[notifCh] = true
subs[sub] = struct{}{}
r.mu.Unlock()
return sub
}

defer func() {
r.mu.Lock()
delete(r.runSubscribers, notifCh)
close(notifCh)
r.mu.Unlock()
}()
func (r *actionRepo) unsubscribe(subs map[*subscription]struct{}, sub *subscription) {
r.mu.Lock()
delete(subs, sub)
close(sub.notify)
r.mu.Unlock()
}

// WatchRunUpdates watches for run updates via LISTEN/NOTIFY
func (r *actionRepo) WatchRunUpdates(ctx context.Context, runID *common.RunIdentifier, updates chan<- *models.Run, errs chan<- error) {
runKey := fmt.Sprintf("%s/%s/%s", runID.Project, runID.Domain, runID.Name)
sub := r.subscribe(r.runSubscribers)
defer r.unsubscribe(r.runSubscribers, sub)

for {
select {
case <-ctx.Done():
return
case notifPayload := <-notifCh:
if notifPayload == runKey {
run, err := r.GetRun(ctx, runID)
if err != nil {
if ctx.Err() != nil {
return
}
errs <- err
return
}
select {
case updates <- run:
case <-ctx.Done():
return
}
case <-sub.resync:
case notifPayload := <-sub.notify:
if notifPayload != runKey {
continue
}
}

run, err := r.GetRun(ctx, runID)
if err != nil {
if ctx.Err() != nil {
return
}
errs <- err
return
}
select {
case updates <- run:
case <-ctx.Done():
return
}
}
}

// WatchAllRunUpdates watches for all run updates via LISTEN/NOTIFY
func (r *actionRepo) WatchAllRunUpdates(ctx context.Context, updates chan<- *models.Run, errs chan<- error) {
notifCh := make(chan string, r.notificationConfig.watchBufferSize)

r.mu.Lock()
r.runSubscribers[notifCh] = true
r.mu.Unlock()

defer func() {
r.mu.Lock()
delete(r.runSubscribers, notifCh)
close(notifCh)
r.mu.Unlock()
}()
sub := r.subscribe(r.runSubscribers)
defer r.unsubscribe(r.runSubscribers, sub)

for {
select {
case <-ctx.Done():
return
case notifPayload := <-notifCh:
case <-sub.resync:
select {
case updates <- nil:
case <-ctx.Done():
return
}
case notifPayload := <-sub.notify:
parts := strings.Split(notifPayload, "/")
if len(parts) != 3 {
logger.Warnf(ctx, "Invalid run notification payload: %s", notifPayload)
Expand Down Expand Up @@ -875,24 +897,20 @@ func (r *actionRepo) WatchAllRunUpdates(ctx context.Context, updates chan<- *mod
// WatchAllActionUpdates watches for all action updates for a run via LISTEN/NOTIFY
func (r *actionRepo) WatchAllActionUpdates(ctx context.Context, runID *common.RunIdentifier, updates chan<- *models.Action, errs chan<- error) {
runPrefix := fmt.Sprintf("%s/%s/%s/", runID.Project, runID.Domain, runID.Name)
notifCh := make(chan string, r.notificationConfig.watchBufferSize)

r.mu.Lock()
r.actionSubscribers[notifCh] = true
r.mu.Unlock()

defer func() {
r.mu.Lock()
delete(r.actionSubscribers, notifCh)
close(notifCh)
r.mu.Unlock()
}()
sub := r.subscribe(r.actionSubscribers)
defer r.unsubscribe(r.actionSubscribers, sub)

for {
select {
case <-ctx.Done():
return
case notifPayload := <-notifCh:
case <-sub.resync:
select {
case updates <- nil:
case <-ctx.Done():
return
}
case notifPayload := <-sub.notify:
// Collect this notification plus any others already queued,
// deduplicating by action name so we only query the DB once
// per action during a burst of updates.
Expand All @@ -904,7 +922,7 @@ func (r *actionRepo) WatchAllActionUpdates(ctx context.Context, runID *common.Ru
drain:
for {
select {
case extra := <-notifCh:
case extra := <-sub.notify:
if len(extra) > len(runPrefix) && extra[:len(runPrefix)] == runPrefix {
pending[extra[len(runPrefix):]] = true
}
Expand Down Expand Up @@ -943,24 +961,15 @@ func (r *actionRepo) WatchAllActionUpdates(ctx context.Context, runID *common.Ru
func (r *actionRepo) WatchActionUpdates(ctx context.Context, actionID *common.ActionIdentifier, updates chan<- *models.Action, errs chan<- error) {
targetPayload := fmt.Sprintf("%s/%s/%s/%s",
actionID.Run.Project, actionID.Run.Domain, actionID.Run.Name, actionID.Name)
notifCh := make(chan string, r.notificationConfig.watchBufferSize)

r.mu.Lock()
r.actionSubscribers[notifCh] = true
r.mu.Unlock()

defer func() {
r.mu.Lock()
delete(r.actionSubscribers, notifCh)
close(notifCh)
r.mu.Unlock()
}()
sub := r.subscribe(r.actionSubscribers)
defer r.unsubscribe(r.actionSubscribers, sub)

for {
select {
case <-ctx.Done():
return
case notifPayload := <-notifCh:
case <-sub.resync:
case notifPayload := <-sub.notify:
if notifPayload != targetPayload {
continue
}
Expand All @@ -969,29 +978,29 @@ func (r *actionRepo) WatchActionUpdates(ctx context.Context, actionID *common.Ac
for {
// Prevent continuous update from the action
select {
case extra := <-notifCh:
case extra := <-sub.notify:
if extra != targetPayload {
continue
}
default:
break drain
}
}
}

action, err := r.GetAction(ctx, actionID)
if err != nil {
if ctx.Err() != nil {
return
}
logger.Errorf(ctx, "Failed to get action from notification: %v", err)
continue
}

select {
case updates <- action:
case <-ctx.Done():
action, err := r.GetAction(ctx, actionID)
if err != nil {
if ctx.Err() != nil {
return
}
logger.Errorf(ctx, "Failed to get action from notification: %v", err)
continue
}

select {
case updates <- action:
case <-ctx.Done():
return
}
}
}
Expand Down Expand Up @@ -1047,16 +1056,25 @@ func (r *actionRepo) processNotifications() {
}
if notif == nil {
// pq sends nil after it reconnects; notifications sent while it was down may be lost.
logger.Infof(context.Background(), "PostgreSQL listener reconnected, resyncing subscribers")
r.mu.RLock()
for sub := range r.runSubscribers {
sub.requestResync()
}
for sub := range r.actionSubscribers {
sub.requestResync()
}
r.mu.RUnlock()
continue
}

switch notif.Channel {
case "run_updates":
// Broadcast to all run subscribers
r.mu.RLock()
for ch := range r.runSubscribers {
for sub := range r.runSubscribers {
select {
case ch <- notif.Extra:
case sub.notify <- notif.Extra:
default:
// Channel full, skip this subscriber
logger.Warnf(context.Background(), "Run subscriber channel full, dropping notification")
Expand All @@ -1067,9 +1085,9 @@ func (r *actionRepo) processNotifications() {
case "action_updates":
// Broadcast to all action subscribers
r.mu.RLock()
for ch := range r.actionSubscribers {
for sub := range r.actionSubscribers {
select {
case ch <- notif.Extra:
case sub.notify <- notif.Extra:
default:
logger.Warnf(context.Background(), "Action subscriber channel full, dropping notification payload=%s", notif.Extra)
}
Expand Down
Loading
Loading