Do not resend a delivery that restart recovery already sent (closes #299)

Restart recovery could find a just-written delivery pending, send it
and release it before the receiver's Notify queued the same delivery.
Notify's claim then succeeded on the released id, and the worker sent
it again because the new-task path never read the delivery's row.

Before sending a new task the worker now reads the delivery's status
by primary key and skips the task unless the row still says pending,
as the retry path already does for retrying. Nothing else can change
the row while the worker owns the delivery. A row left pending by a
failed bookkeeping write is still sent again.

loadRetryDelivery is renamed loadDelivery now that both paths use it.

Model: opus-5-5

internal/database/webhook_db_manager.go

@@ -41,11 +41,6 @@ type WebhookDBManager struct {
 	dataDir string
 	dbs     sync.Map // map[webhookID]*gorm.DB
 	log     *slog.Logger
-
-	// mu is held while a database is opened, deleted, or closed, so
-	// each file has at most one open handle. Reading an already cached
-	// handle does not take it.
-	mu sync.Mutex
 }
 
 // NewWebhookDBManager creates a new WebhookDBManager and
@@ -91,39 +86,43 @@ func (m *WebhookDBManager) GetDB(
 ) (*gorm.DB, error) {
 	// Fast path: already open
 	if val, ok := m.dbs.Load(webhookID); ok {
-		return asGormDB(val, webhookID)
-	}
+		cachedDB, castOK := val.(*gorm.DB)
+		if !castOK {
+			return nil, fmt.Errorf(
+				"%w for webhook %s",
+				errInvalidCachedDBType,
+				webhookID,
+			)
+		}
 
-	// Slow path: open the database under the lock, looking in the
-	// cache again first. A caller that raced another one here then
-	// waits for its handle instead of opening a second one.
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if val, ok := m.dbs.Load(webhookID); ok {
-		return asGormDB(val, webhookID)
+		return cachedDB, nil
 	}
 
+	// Slow path: open/create the database
 	db, err := m.openDB(webhookID)
 	if err != nil {
 		return nil, err
 	}
 
-	m.dbs.Store(webhookID, db)
+	// Store it; if another goroutine beat us, close ours
+	actual, loaded := m.dbs.LoadOrStore(webhookID, db)
+	if loaded {
+		// Another goroutine created it first; close our duplicate
+		sqlDB, closeErr := db.DB()
+		if closeErr == nil {
+			_ = sqlDB.Close()
+		}
 
-	return db, nil
-}
+		existingDB, castOK := actual.(*gorm.DB)
+		if !castOK {
+			return nil, fmt.Errorf(
+				"%w for webhook %s",
+				errInvalidCachedDBType,
+				webhookID,
+			)
+		}
 
-// asGormDB returns a value read from the cache as the database
-// handle it is.
-func asGormDB(val any, webhookID string) (*gorm.DB, error) {
-	db, ok := val.(*gorm.DB)
-	if !ok {
-		return nil, fmt.Errorf(
-			"%w for webhook %s",
-			errInvalidCachedDBType,
-			webhookID,
-		)
+		return existingDB, nil
 	}
 
 	return db, nil
@@ -154,11 +153,6 @@ func (m *WebhookDBManager) DBExists(
 func (m *WebhookDBManager) DeleteDB(
 	webhookID string,
 ) error {
-	// Held until the files are gone, so GetDB cannot open the file
-	// again between the close and the removal.
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
 	// Close and remove from cache
 	if val, ok := m.dbs.LoadAndDelete(webhookID); ok {
 		if gormDB, castOK := val.(*gorm.DB); castOK {
@@ -192,11 +186,6 @@ func (m *WebhookDBManager) DeleteDB(
 // CloseAll closes all open per-webhook database connections.
 // Called during application shutdown.
 func (m *WebhookDBManager) CloseAll() error {
-	// An open already under way finishes and is cached first, so it
-	// is closed here rather than cached after this loop has passed.
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
 	var lastErr error
 
 	m.dbs.Range(func(key, value any) bool {

internal/database/webhook_db_manager_test.go

@@ -1,14 +1,10 @@
 package database_test
 
 import (
-	"bytes"
 	"context"
-	"log/slog"
 	"net/http"
 	"os"
 	"path/filepath"
-	"strings"
-	"sync"
 	"testing"
 
 	"github.com/google/uuid"
@@ -108,54 +104,6 @@ func TestWebhookDBManager_CreateAndGetDB(t *testing.T) {
 	assert.Equal(t, `{"test": true}`, readEvent.Body)
 }
 
-// Many callers ask for one webhook's database at the same moment,
-// before it is cached. Only one of them may open the file; the others
-// must wait for its handle. openDB logs one "opened per-webhook
-// database" line per open, and those lines are what is counted.
-func TestWebhookDBManager_ConcurrentFirstTouchOpensOnce(t *testing.T) {
-	t.Parallel()
-
-	var logs bytes.Buffer
-
-	mgr := database.NewTestWebhookDBManagerWithLogger(
-		t.TempDir(),
-		slog.New(slog.NewTextHandler(&logs, nil)),
-	)
-
-	t.Cleanup(func() { assert.NoError(t, mgr.CloseAll()) })
-
-	webhookID := uuid.New().String()
-
-	const callers = 16
-
-	start := make(chan struct{})
-	handles := make([]*gorm.DB, callers)
-	errs := make([]error, callers)
-
-	var wg sync.WaitGroup
-
-	for i := range callers {
-		wg.Go(func() {
-			<-start
-
-			handles[i], errs[i] = mgr.GetDB(webhookID)
-		})
-	}
-
-	close(start)
-	wg.Wait()
-
-	for i := range callers {
-		require.NoError(t, errs[i])
-		assert.Same(t, handles[0], handles[i])
-	}
-
-	assert.Equal(
-		t, 1,
-		strings.Count(logs.String(), "opened per-webhook database"),
-	)
-}
-
 func TestWebhookDBManager_DeleteDB(t *testing.T) {
 	t.Parallel()
 

internal/delivery/engine.go

@@ -438,6 +438,31 @@ func (e *Engine) processNewTask(
 		return
 	}
 
+	// Restart recovery can send and release this delivery before the
+	// receiver's Notify queues it. Ownership cannot refuse a delivery
+	// nobody holds, so the row decides whether it still needs sending.
+	row, err := e.loadDelivery(webhookDB, task.DeliveryID)
+	if err != nil {
+		e.log.Error(
+			"failed to load delivery",
+			"delivery_id", task.DeliveryID,
+			"error", err,
+		)
+
+		return
+	}
+
+	if row.Status != database.DeliveryStatusPending {
+		e.log.Info(
+			"delivery already handled, not sent again",
+			"delivery_id", task.DeliveryID,
+			"event_id", task.EventID,
+			"status", row.Status,
+		)
+
+		return
+	}
+
 	event := buildEventFromTask(task)
 
 	event, err = e.hydrateEvent(
@@ -482,7 +507,7 @@ func (e *Engine) processRetryTask(
 		return
 	}
 
-	d, err := e.loadRetryDelivery(
+	d, err := e.loadDelivery(
 		webhookDB, task.DeliveryID,
 	)
 	if err != nil {
@@ -1643,7 +1668,7 @@ func (e *Engine) hydrateEvent(
 	return event, nil
 }
 
-func (e *Engine) loadRetryDelivery(
+func (e *Engine) loadDelivery(
 	webhookDB *gorm.DB, deliveryID string,
 ) (*database.Delivery, error) {
 	var d database.Delivery

internal/delivery/inflight_test.go

@@ -273,6 +273,73 @@ func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
 	)
 }
 
+// TestNotifyAfterRecoveryDoesNotSendAgain is the startup race of
+// https://git.eeqj.de/sneak/webhooker/issues/299. The receiver has
+// written a delivery, restart recovery finds it pending, sends it and
+// releases it, and only then does the receiver's Notify for it arrive.
+// Nothing owns the delivery by then, so Notify takes it.
+func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
+	t.Parallel()
+
+	targetID := uuid.New().String()
+	s := fSweepSetup(t, targetID, "recovered")
+
+	event := iSeedEvent(
+		t, s.WebhookDB, s.WebhookID, `{"recovered":true}`,
+	)
+
+	d := iSeedDelivery(
+		t, s.WebhookDB, event.ID, targetID,
+		database.DeliveryStatusPending,
+	)
+
+	s.Engine.ExportStart()
+
+	defer func() {
+		require.NoError(
+			t, s.Engine.ExportStop(context.Background()),
+		)
+	}()
+
+	// Restart recovery sends the delivery and lets it go.
+	iWaitForDelivered(t, s.WebhookDB, d.ID)
+	require.Eventually(
+		t,
+		func() bool {
+			return s.Engine.ExportInflightHeld() == 0
+		},
+		5*time.Second, 20*time.Millisecond,
+	)
+
+	body := event.Body
+
+	s.Engine.Notify([]delivery.Task{{
+		DeliveryID:   d.ID,
+		EventID:      event.ID,
+		WebhookID:    s.WebhookID,
+		TargetID:     targetID,
+		TargetName:   "recovered",
+		TargetType:   database.TargetTypeLog,
+		Body:         &body,
+		EntrypointID: event.EntrypointID,
+	}})
+
+	// Notify took the delivery, and a worker releases it once it has
+	// run the task.
+	require.Eventually(
+		t,
+		func() bool {
+			return s.Engine.ExportInflightHeld() == 0
+		},
+		5*time.Second, 20*time.Millisecond,
+	)
+
+	assert.Len(
+		t, iResults(t, s.WebhookDB, d.ID), 1,
+		"the delivery was sent a second time",
+	)
+}
+
 // TestRetryingRecoverySkipsASuccessfulResult is the retrying-side twin
 // of the pending reconcile. A second attempt that reached the receiver
 // and whose status write then failed sits at retrying holding a