Open each event database once when callers race (closes #291)
check / check (push) Successful in 3m44s

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
This commit is contained in:
2026-09-29 01:23:16 +00:00
parent aeeeca5ea1
commit c8f81d3c46
2 changed files with 90 additions and 27 deletions
+32 -21
View File
@@ -41,6 +41,11 @@ type WebhookDBManager struct {
dataDir string dataDir string
dbs sync.Map // map[webhookID]*gorm.DB dbs sync.Map // map[webhookID]*gorm.DB
log *slog.Logger 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 // NewWebhookDBManager creates a new WebhookDBManager and
@@ -86,35 +91,34 @@ func (m *WebhookDBManager) GetDB(
) (*gorm.DB, error) { ) (*gorm.DB, error) {
// Fast path: already open // Fast path: already open
if val, ok := m.dbs.Load(webhookID); ok { if val, ok := m.dbs.Load(webhookID); ok {
cachedDB, castOK := val.(*gorm.DB) return asGormDB(val, webhookID)
if !castOK {
return nil, fmt.Errorf(
"%w for webhook %s",
errInvalidCachedDBType,
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) db, err := m.openDB(webhookID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
// Store it; if another goroutine beat us, close ours m.dbs.Store(webhookID, db)
actual, loaded := m.dbs.LoadOrStore(webhookID, db)
if loaded { return db, nil
// Another goroutine created it first; close our duplicate
sqlDB, closeErr := db.DB()
if closeErr == nil {
_ = sqlDB.Close()
} }
existingDB, castOK := actual.(*gorm.DB) // asGormDB returns a value read from the cache as the database
if !castOK { // handle it is.
func asGormDB(val any, webhookID string) (*gorm.DB, error) {
db, ok := val.(*gorm.DB)
if !ok {
return nil, fmt.Errorf( return nil, fmt.Errorf(
"%w for webhook %s", "%w for webhook %s",
errInvalidCachedDBType, errInvalidCachedDBType,
@@ -122,9 +126,6 @@ func (m *WebhookDBManager) GetDB(
) )
} }
return existingDB, nil
}
return db, nil return db, nil
} }
@@ -153,6 +154,11 @@ func (m *WebhookDBManager) DBExists(
func (m *WebhookDBManager) DeleteDB( func (m *WebhookDBManager) DeleteDB(
webhookID string, webhookID string,
) error { ) 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 // Close and remove from cache
if val, ok := m.dbs.LoadAndDelete(webhookID); ok { if val, ok := m.dbs.LoadAndDelete(webhookID); ok {
if gormDB, castOK := val.(*gorm.DB); castOK { if gormDB, castOK := val.(*gorm.DB); castOK {
@@ -186,6 +192,11 @@ func (m *WebhookDBManager) DeleteDB(
// CloseAll closes all open per-webhook database connections. // CloseAll closes all open per-webhook database connections.
// Called during application shutdown. // Called during application shutdown.
func (m *WebhookDBManager) CloseAll() error { 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 var lastErr error
m.dbs.Range(func(key, value any) bool { m.dbs.Range(func(key, value any) bool {
@@ -1,10 +1,14 @@
package database_test package database_test
import ( import (
"bytes"
"context" "context"
"log/slog"
"net/http" "net/http"
"os" "os"
"path/filepath" "path/filepath"
"strings"
"sync"
"testing" "testing"
"github.com/google/uuid" "github.com/google/uuid"
@@ -104,6 +108,54 @@ func TestWebhookDBManager_CreateAndGetDB(t *testing.T) {
assert.Equal(t, `{"test": true}`, readEvent.Body) 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) { func TestWebhookDBManager_DeleteDB(t *testing.T) {
t.Parallel() t.Parallel()