triggers loop

Author: pierre-b

CHANGELOG.md

@@ -2,9 +2,17 @@
 
 All notable changes to this project will be documented in this file.
 
-## [30.4] - 2026-05-18
+## [31.0] - 2026-05-19
 
-- **Fix**: Workspace database pool no longer falsely evicted when the caller's HTTP request context expires mid-Ping during broadcast load. Previously, a single caller-context cancellation could close the cached pool, causing every other in-flight worker (segment queue, webhook delivery, email send) to fail with `sql: database is closed` on its next operation. The pool health check now uses an isolated, sub-second context.
+### Database Schema Changes
+
+- Migration v31.0 updates the `queue_contact_for_segment_recomputation` trigger function on every workspace database to short-circuit when the inserted `contact_timeline` row is itself a segment membership event (`kind IN ('segment.joined', 'segment.left')`).
+
+### Fixes
+
+- **Fix**: `queue_contact_for_segment_recomputation` trigger no longer re-enqueues contacts when the inserted `contact_timeline` event is itself a segment membership change (`segment.joined`/`segment.left`). Removes a self-loop where every membership write re-queued the same contact.
+- **Fix**: Recurring tasks dispatched via HTTP now write `timeout_after` in UTC. The column is `TIMESTAMP WITHOUT TIME ZONE` and the scheduler compares it against `time.Now().UTC()`; on non-UTC hosts the local-time value caused the task to appear "still running" for the host's UTC offset. Same fix applied to the broadcast-pause `next_run_after`.
+- **Fix**: `GetWorkspaceConnection`'s pool health check now uses an isolated context for `pool.PingContext` instead of the caller's. A caller-context cancellation no longer triggers pool eviction.
 
 ## [30.3] - 2026-05-14
 

config/config.go

@@ -15,7 +15,7 @@ import (
 	"github.com/spf13/viper"
 )
 
-const VERSION = "30.4"
+const VERSION = "31.0"
 
 type Config struct {
 	Server              ServerConfig

internal/database/init.go

@@ -706,6 +706,18 @@ func InitializeWorkspaceDatabase(db *sql.DB) error {
 		`CREATE OR REPLACE FUNCTION queue_contact_for_segment_recomputation()
 		RETURNS TRIGGER AS $$
 		BEGIN
+			-- Skip re-queue when the timeline event is itself a segment
+			-- membership change. The queue worker writes contact_segments,
+			-- which fires track_contact_segment_changes (inserts a
+			-- contact_timeline row with kind='segment.joined'/'segment.left'),
+			-- which would re-enter this function and re-queue the same
+			-- contact for re-processing — a self-loop that also contends
+			-- with concurrent open-tracking writes on the contact_segment_queue
+			-- (email) row lock and prevents batches from completing.
+			IF NEW.kind IN ('segment.joined', 'segment.left') THEN
+				RETURN NEW;
+			END IF;
+
 			-- Queue the contact for segment recomputation
 			INSERT INTO contact_segment_queue (email, queued_at)
 			VALUES (NEW.email, CURRENT_TIMESTAMP)

internal/http/task_handler.go

@@ -237,8 +237,13 @@ func (h *TaskHandler) ExecuteTask(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Calculate timeout based on task's MaxRuntime
-	timeoutAt := time.Now().Add(time.Duration(task.MaxRuntime) * time.Second)
+	// Calculate timeout based on task's MaxRuntime. UTC is mandatory: the
+	// tasks.timeout_after column is TIMESTAMP WITHOUT TIME ZONE, so a local
+	// time.Now() on a non-UTC server would write a literal that, when later
+	// compared against time.Now().UTC() in GetNextBatch, leaves the task
+	// "still running" for the duration of the server's UTC offset (e.g. +2h
+	// in CEST) and the recurring task is never re-picked within that window.
+	timeoutAt := time.Now().UTC().Add(time.Duration(task.MaxRuntime) * time.Second)
 
 	if err := h.taskService.ExecuteTask(r.Context(), executeRequest.WorkspaceID, executeRequest.ID, timeoutAt); err != nil {
 		// Handle different error types with appropriate status codes

internal/migrations/manager_test.go

@@ -541,7 +541,7 @@ func TestManager_RunMigrations_AdditionalCoverage(t *testing.T) {
 
 		// Mock GetCurrentDBVersion to return the latest migrated version (up to date)
 		mock.ExpectQuery("SELECT value FROM settings WHERE key = 'db_version'").
-			WillReturnRows(sqlmock.NewRows([]string{"value"}).AddRow("30"))
+			WillReturnRows(sqlmock.NewRows([]string{"value"}).AddRow("31"))
 
 		err = manager.RunMigrations(context.Background(), cfg, db)
 

internal/migrations/v31.go

@@ -0,0 +1,84 @@
+package migrations
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/Notifuse/notifuse/config"
+	"github.com/Notifuse/notifuse/internal/domain"
+)
+
+// V31Migration breaks the segment-recomputation self-loop.
+//
+// The previous queue_contact_for_segment_recomputation() unconditionally
+// upserted into contact_segment_queue whenever any row was inserted into
+// contact_timeline. That included rows produced by track_contact_segment_changes
+// (kind='segment.joined' / 'segment.left'), which fire as the queue worker
+// itself writes contact_segments. The effect was twofold:
+//
+//  1. Every membership write re-queued the same contact for re-processing
+//     after the 15s debounce — pure wasted work.
+//  2. The worker's INSERT contact_segments transaction ended up contending
+//     with concurrent /opens and delivery-webhook writers on the same
+//     (email) row lock in contact_segment_queue. Under broadcast load this
+//     stalled the worker for >10s per batch and caused the queue to
+//     accumulate without draining.
+//
+// The function now short-circuits when the timeline event is itself a
+// segment membership change, leaving every other event path unchanged.
+type V31Migration struct{}
+
+func (m *V31Migration) GetMajorVersion() float64 {
+	return 31.0
+}
+
+func (m *V31Migration) HasSystemUpdate() bool {
+	return false
+}
+
+func (m *V31Migration) HasWorkspaceUpdate() bool {
+	return true
+}
+
+func (m *V31Migration) ShouldRestartServer() bool {
+	return false
+}
+
+func (m *V31Migration) UpdateSystem(ctx context.Context, cfg *config.Config, db DBExecutor) error {
+	return nil
+}
+
+func (m *V31Migration) UpdateWorkspace(ctx context.Context, cfg *config.Config, workspace *domain.Workspace, db DBExecutor) error {
+	_, err := db.ExecContext(ctx, `
+		CREATE OR REPLACE FUNCTION queue_contact_for_segment_recomputation()
+		RETURNS TRIGGER AS $$
+		BEGIN
+			-- Skip re-queue when the timeline event is itself a segment
+			-- membership change. The queue worker writes contact_segments,
+			-- which fires track_contact_segment_changes (inserts a
+			-- contact_timeline row with kind='segment.joined'/'segment.left'),
+			-- which would re-enter this function and re-queue the same
+			-- contact for re-processing — a self-loop that also contends
+			-- with concurrent open-tracking writes on the contact_segment_queue
+			-- (email) row lock and prevents batches from completing.
+			IF NEW.kind IN ('segment.joined', 'segment.left') THEN
+				RETURN NEW;
+			END IF;
+
+			INSERT INTO contact_segment_queue (email, queued_at)
+			VALUES (NEW.email, CURRENT_TIMESTAMP)
+			ON CONFLICT (email) DO UPDATE SET queued_at = EXCLUDED.queued_at;
+			RETURN NEW;
+		END;
+		$$ LANGUAGE plpgsql;
+	`)
+	if err != nil {
+		return fmt.Errorf("failed to update queue_contact_for_segment_recomputation function for workspace %s: %w", workspace.ID, err)
+	}
+
+	return nil
+}
+
+func init() {
+	Register(&V31Migration{})
+}

internal/migrations/v31_test.go

@@ -0,0 +1,80 @@
+package migrations
+
+import (
+	"context"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/Notifuse/notifuse/config"
+	"github.com/Notifuse/notifuse/internal/domain"
+)
+
+func TestV31Migration_GetMajorVersion(t *testing.T) {
+	m := &V31Migration{}
+	assert.Equal(t, 31.0, m.GetMajorVersion())
+}
+
+func TestV31Migration_HasSystemUpdate(t *testing.T) {
+	m := &V31Migration{}
+	assert.False(t, m.HasSystemUpdate())
+}
+
+func TestV31Migration_HasWorkspaceUpdate(t *testing.T) {
+	m := &V31Migration{}
+	assert.True(t, m.HasWorkspaceUpdate())
+}
+
+func TestV31Migration_ShouldRestartServer(t *testing.T) {
+	m := &V31Migration{}
+	assert.False(t, m.ShouldRestartServer())
+}
+
+func TestV31Migration_UpdateSystem_NoOp(t *testing.T) {
+	m := &V31Migration{}
+	// System-side has nothing to do — just verify it returns nil cleanly.
+	assert.NoError(t, m.UpdateSystem(context.Background(), &config.Config{}, nil))
+}
+
+func TestV31Migration_UpdateWorkspace_Success(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	defer db.Close()
+
+	mock.ExpectExec(`CREATE OR REPLACE FUNCTION queue_contact_for_segment_recomputation`).
+		WillReturnResult(sqlmock.NewResult(0, 0))
+
+	m := &V31Migration{}
+	err = m.UpdateWorkspace(context.Background(), &config.Config{},
+		&domain.Workspace{ID: "ws_test"}, db)
+	assert.NoError(t, err)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestV31Migration_UpdateWorkspace_Error(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	defer db.Close()
+
+	mock.ExpectExec(`CREATE OR REPLACE FUNCTION queue_contact_for_segment_recomputation`).
+		WillReturnError(assert.AnError)
+
+	m := &V31Migration{}
+	err = m.UpdateWorkspace(context.Background(), &config.Config{},
+		&domain.Workspace{ID: "ws_test"}, db)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "failed to update queue_contact_for_segment_recomputation")
+	assert.Contains(t, err.Error(), "ws_test")
+}
+
+func TestV31Migration_Registered(t *testing.T) {
+	// Verify init() registered the migration so the runner picks it up.
+	for _, m := range GetRegisteredMigrations() {
+		if m.GetMajorVersion() == 31.0 {
+			return
+		}
+	}
+	t.Fatal("V31Migration not registered")
+}

internal/service/task_service.go

@@ -990,8 +990,9 @@ func (s *TaskService) handleBroadcastPaused(ctx context.Context, payload domain.
 		return
 	}
 
-	// Pause the task
-	nextRunAfter := time.Now().Add(24 * time.Hour) // Pause for 24 hours
+	// Pause the task. UTC is mandatory because tasks.next_run_after is
+	// TIMESTAMP WITHOUT TIME ZONE — see task_handler.go timeoutAt comment.
+	nextRunAfter := time.Now().UTC().Add(24 * time.Hour) // Pause for 24 hours
 	tracing.AddAttribute(ctx, "next_run_after", nextRunAfter.Format(time.RFC3339))
 
 	if err := s.repo.MarkAsPaused(ctx, payload.WorkspaceID, task.ID, nextRunAfter, task.Progress, task.State); err != nil {