Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
8 changes: 4 additions & 4 deletions apps/backend/internal/orchestrator/ci_automation_attempt.go
Original file line number Diff line number Diff line change
Expand Up @@ -103,10 +103,10 @@ func (s *Service) reconcileOrphanedCIAutoFixQueueEntries(
continue
}
}
if ackErr := s.messageQueue.AcknowledgeQueued(
context.WithoutCancel(ctx), &entry,
); ackErr != nil {
return fmt.Errorf("remove orphaned CI auto-fix queue entry %q: %w", entry.ID, ackErr)
if removeErr := s.messageQueue.RemoveEntry(
context.WithoutCancel(ctx), entry.SessionID, entry.ID,
); removeErr != nil && !errors.Is(removeErr, messagequeue.ErrEntryNotFound) {
return fmt.Errorf("remove orphaned CI auto-fix queue entry %q: %w", entry.ID, removeErr)
}
}
return nil
Expand Down
9 changes: 1 addition & 8 deletions apps/backend/internal/task/repository/sqlite/session.go
Original file line number Diff line number Diff line change
Expand Up @@ -2920,14 +2920,7 @@ func unmarshalSessionSnapshots(
// in this task transaction. Some repository unit tests intentionally omit the
// message queue schema; production databases always include it.
func (r *Repository) queueSessionLockTablePresent(ctx context.Context) (bool, error) {
var present bool
var err error
if dialect.IsPostgres(r.db.DriverName()) {
err = r.db.GetContext(ctx, &present, `SELECT to_regclass('queue_session_locks') IS NOT NULL`)
} else {
err = r.db.GetContext(ctx, &present, `SELECT EXISTS (SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'queue_session_locks')`)
}
return present, err
return db.TableExistsContext(ctx, r.db, "queue_session_locks")
}

// DeleteTaskSession deletes the exact session incarnation and its pending queue
Expand Down
Loading