Open each event database once when callers race (closes #291)

GetDB opened the database on a cache miss and then tried to cache it,
so callers racing on a webhook's first use could each open the file,
and the losers closed their copies. On a new file the parallel opens
also create its tables at the same time, and one caller can fail with
"table already exists".

A mutex now covers the open: GetDB looks in the cache again under it,
then opens and caches. DeleteDB and CloseAll take the same mutex, so
neither runs while an open is under way. Reading an already cached
database takes no lock.

The new test starts many callers on one webhook at once and checks
that exactly one open happened.

Model: opus-5-5

internal/database/webhook_db_manager.go

@@ -41,6 +41,11 @@ 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
@@ -86,43 +91,39 @@ func (m *WebhookDBManager) GetDB(
 ) (*gorm.DB, error) {
 	// Fast path: already open
 	if val, ok := m.dbs.Load(webhookID); ok {
-		cachedDB, castOK := val.(*gorm.DB)
-		if !castOK {
-			return nil, fmt.Errorf(
-				"%w for webhook %s",
-				errInvalidCachedDBType,
-				webhookID,
-			)
-		}
+		return asGormDB(val, webhookID)
+	}
 
-		return cachedDB, nil
+	// 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)
 	}
 
-	// Slow path: open/create the database
 	db, err := m.openDB(webhookID)
 	if err != nil {
 		return nil, err
 	}
 
-	// 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()
-		}
+	m.dbs.Store(webhookID, db)
 
-		existingDB, castOK := actual.(*gorm.DB)
-		if !castOK {
-			return nil, fmt.Errorf(
-				"%w for webhook %s",
-				errInvalidCachedDBType,
-				webhookID,
-			)
-		}
+	return db, nil
+}
 
-		return existingDB, nil
+// 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 db, nil
@@ -153,6 +154,11 @@ 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 {
@@ -186,6 +192,11 @@ 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,10 +1,14 @@
 package database_test
 
 import (
+	"bytes"
 	"context"
+	"log/slog"
 	"net/http"
 	"os"
 	"path/filepath"
+	"strings"
+	"sync"
 	"testing"
 
 	"github.com/google/uuid"
@@ -104,6 +108,54 @@ 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,31 +438,6 @@ 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(
@@ -507,7 +482,7 @@ func (e *Engine) processRetryTask(
 		return
 	}
 
-	d, err := e.loadDelivery(
+	d, err := e.loadRetryDelivery(
 		webhookDB, task.DeliveryID,
 	)
 	if err != nil {
@@ -1668,7 +1643,7 @@ func (e *Engine) hydrateEvent(
 	return event, nil
 }
 
-func (e *Engine) loadDelivery(
+func (e *Engine) loadRetryDelivery(
 	webhookDB *gorm.DB, deliveryID string,
 ) (*database.Delivery, error) {
 	var d database.Delivery

internal/delivery/inflight_test.go

@@ -273,73 +273,6 @@ 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