clawbot
978eb01b29
check / check (push) Successful in 3m48s
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
424 lines
9.2 KiB
Go
424 lines
9.2 KiB
Go
package database_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"log/slog"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"go.uber.org/fx/fxtest"
|
|
"gorm.io/gorm"
|
|
"sneak.berlin/go/webhooker/internal/config"
|
|
"sneak.berlin/go/webhooker/internal/database"
|
|
"sneak.berlin/go/webhooker/internal/globals"
|
|
"sneak.berlin/go/webhooker/internal/logger"
|
|
)
|
|
|
|
func setupTestWebhookDBManager(
|
|
t *testing.T,
|
|
) (*database.WebhookDBManager, *fxtest.Lifecycle) {
|
|
t.Helper()
|
|
|
|
lc := fxtest.NewLifecycle(t)
|
|
|
|
g := &globals.Globals{
|
|
Appname: testAppname,
|
|
Version: testVersion,
|
|
}
|
|
|
|
l, err := logger.New(
|
|
lc,
|
|
logger.LoggerParams{Globals: g},
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
dataDir := filepath.Join(t.TempDir(), "events")
|
|
|
|
cfg := &config.Config{
|
|
DataDir: dataDir,
|
|
}
|
|
|
|
mgr, err := database.NewWebhookDBManager(
|
|
lc,
|
|
database.WebhookDBManagerParams{
|
|
Config: cfg,
|
|
Logger: l,
|
|
},
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
return mgr, lc
|
|
}
|
|
|
|
func TestWebhookDBManager_CreateAndGetDB(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
mgr, lc := setupTestWebhookDBManager(t)
|
|
ctx := context.Background()
|
|
require.NoError(t, lc.Start(ctx))
|
|
|
|
defer func() { require.NoError(t, lc.Stop(ctx)) }()
|
|
|
|
webhookID := uuid.New().String()
|
|
|
|
// DB should not exist yet
|
|
assert.False(t, mgr.DBExists(webhookID))
|
|
|
|
// Create the DB
|
|
err := mgr.CreateDB(webhookID)
|
|
require.NoError(t, err)
|
|
|
|
// DB file should now exist
|
|
assert.True(t, mgr.DBExists(webhookID))
|
|
|
|
// Get the DB again (should use cached connection)
|
|
db, err := mgr.GetDB(webhookID)
|
|
require.NoError(t, err)
|
|
require.NotNil(t, db)
|
|
|
|
// Verify we can write an event
|
|
event := &database.Event{
|
|
WebhookID: webhookID,
|
|
EntrypointID: uuid.New().String(),
|
|
Method: http.MethodPost,
|
|
Headers: `{"Content-Type":["application/json"]}`,
|
|
Body: `{"test": true}`,
|
|
ContentType: testContentType,
|
|
}
|
|
require.NoError(t, db.Create(event).Error)
|
|
assert.NotEmpty(t, event.ID)
|
|
|
|
// Verify we can read it back
|
|
var readEvent database.Event
|
|
|
|
require.NoError(
|
|
t,
|
|
db.First(&readEvent, "id = ?", event.ID).Error,
|
|
)
|
|
assert.Equal(t, webhookID, readEvent.WebhookID)
|
|
assert.Equal(t, http.MethodPost, readEvent.Method)
|
|
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()
|
|
|
|
mgr, lc := setupTestWebhookDBManager(t)
|
|
ctx := context.Background()
|
|
require.NoError(t, lc.Start(ctx))
|
|
|
|
defer func() { require.NoError(t, lc.Stop(ctx)) }()
|
|
|
|
webhookID := uuid.New().String()
|
|
|
|
// Create the DB and write some data
|
|
require.NoError(t, mgr.CreateDB(webhookID))
|
|
|
|
db, err := mgr.GetDB(webhookID)
|
|
require.NoError(t, err)
|
|
|
|
event := &database.Event{
|
|
WebhookID: webhookID,
|
|
EntrypointID: uuid.New().String(),
|
|
Method: http.MethodPost,
|
|
Body: `{"test": true}`,
|
|
ContentType: testContentType,
|
|
}
|
|
require.NoError(t, db.Create(event).Error)
|
|
|
|
// Delete the DB
|
|
require.NoError(t, mgr.DeleteDB(webhookID))
|
|
|
|
// File should no longer exist
|
|
assert.False(t, mgr.DBExists(webhookID))
|
|
|
|
// Verify the file is actually gone from disk
|
|
dbPath := mgr.DBPath(webhookID)
|
|
|
|
_, err = os.Stat(dbPath)
|
|
assert.True(t, os.IsNotExist(err))
|
|
}
|
|
|
|
func TestWebhookDBManager_LazyCreation(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
mgr, lc := setupTestWebhookDBManager(t)
|
|
ctx := context.Background()
|
|
require.NoError(t, lc.Start(ctx))
|
|
|
|
defer func() { require.NoError(t, lc.Stop(ctx)) }()
|
|
|
|
webhookID := uuid.New().String()
|
|
|
|
// GetDB should lazily create the database
|
|
db, err := mgr.GetDB(webhookID)
|
|
require.NoError(t, err)
|
|
require.NotNil(t, db)
|
|
|
|
// File should now exist
|
|
assert.True(t, mgr.DBExists(webhookID))
|
|
}
|
|
|
|
func TestWebhookDBManager_DeliveryWorkflow(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
mgr, lc := setupTestWebhookDBManager(t)
|
|
ctx := context.Background()
|
|
require.NoError(t, lc.Start(ctx))
|
|
|
|
defer func() { require.NoError(t, lc.Stop(ctx)) }()
|
|
|
|
webhookID := uuid.New().String()
|
|
targetID := uuid.New().String()
|
|
|
|
db, err := mgr.GetDB(webhookID)
|
|
require.NoError(t, err)
|
|
|
|
event, delivery := seedDeliveryWorkflow(
|
|
t, db, webhookID, targetID,
|
|
)
|
|
|
|
verifyPendingDeliveries(t, db, event)
|
|
completeDelivery(t, db, delivery)
|
|
verifyNoPending(t, db)
|
|
}
|
|
|
|
func seedDeliveryWorkflow(
|
|
t *testing.T,
|
|
db *gorm.DB,
|
|
webhookID, targetID string,
|
|
) (*database.Event, *database.Delivery) {
|
|
t.Helper()
|
|
|
|
event := &database.Event{
|
|
WebhookID: webhookID,
|
|
EntrypointID: uuid.New().String(),
|
|
Method: http.MethodPost,
|
|
Headers: `{"Content-Type":["application/json"]}`,
|
|
Body: `{"payload": "test"}`,
|
|
ContentType: testContentType,
|
|
}
|
|
require.NoError(t, db.Create(event).Error)
|
|
|
|
delivery := &database.Delivery{
|
|
EventID: event.ID,
|
|
TargetID: targetID,
|
|
Status: database.DeliveryStatusPending,
|
|
}
|
|
require.NoError(t, db.Create(delivery).Error)
|
|
|
|
return event, delivery
|
|
}
|
|
|
|
func verifyPendingDeliveries(
|
|
t *testing.T,
|
|
db *gorm.DB,
|
|
event *database.Event,
|
|
) {
|
|
t.Helper()
|
|
|
|
var pending []database.Delivery
|
|
|
|
require.NoError(
|
|
t,
|
|
db.Where(
|
|
"status = ?",
|
|
database.DeliveryStatusPending,
|
|
).Preload("Event").Find(&pending).Error,
|
|
)
|
|
require.Len(t, pending, 1)
|
|
assert.Equal(t, event.ID, pending[0].EventID)
|
|
assert.Equal(t, http.MethodPost, pending[0].Event.Method)
|
|
}
|
|
|
|
func completeDelivery(
|
|
t *testing.T,
|
|
db *gorm.DB,
|
|
delivery *database.Delivery,
|
|
) {
|
|
t.Helper()
|
|
|
|
result := &database.DeliveryResult{
|
|
DeliveryID: delivery.ID,
|
|
AttemptNum: 1,
|
|
Success: true,
|
|
StatusCode: 200,
|
|
Duration: 42,
|
|
}
|
|
require.NoError(t, db.Create(result).Error)
|
|
|
|
require.NoError(
|
|
t,
|
|
db.Model(delivery).Update(
|
|
"status",
|
|
database.DeliveryStatusDelivered,
|
|
).Error,
|
|
)
|
|
}
|
|
|
|
func verifyNoPending(
|
|
t *testing.T,
|
|
db *gorm.DB,
|
|
) {
|
|
t.Helper()
|
|
|
|
var stillPending []database.Delivery
|
|
|
|
require.NoError(
|
|
t,
|
|
db.Where(
|
|
"status = ?",
|
|
database.DeliveryStatusPending,
|
|
).Find(&stillPending).Error,
|
|
)
|
|
assert.Empty(t, stillPending)
|
|
}
|
|
|
|
func TestWebhookDBManager_MultipleWebhooks(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
mgr, lc := setupTestWebhookDBManager(t)
|
|
ctx := context.Background()
|
|
require.NoError(t, lc.Start(ctx))
|
|
|
|
defer func() { require.NoError(t, lc.Stop(ctx)) }()
|
|
|
|
webhook1 := uuid.New().String()
|
|
webhook2 := uuid.New().String()
|
|
|
|
// Create DBs for two webhooks
|
|
require.NoError(t, mgr.CreateDB(webhook1))
|
|
require.NoError(t, mgr.CreateDB(webhook2))
|
|
|
|
db1, err := mgr.GetDB(webhook1)
|
|
require.NoError(t, err)
|
|
|
|
db2, err := mgr.GetDB(webhook2)
|
|
require.NoError(t, err)
|
|
|
|
// Write events to each webhook's DB
|
|
event1 := &database.Event{
|
|
WebhookID: webhook1,
|
|
EntrypointID: uuid.New().String(),
|
|
Method: http.MethodPost,
|
|
Body: `{"webhook": 1}`,
|
|
ContentType: testContentType,
|
|
}
|
|
event2 := &database.Event{
|
|
WebhookID: webhook2,
|
|
EntrypointID: uuid.New().String(),
|
|
Method: http.MethodPut,
|
|
Body: `{"webhook": 2}`,
|
|
ContentType: testContentType,
|
|
}
|
|
|
|
require.NoError(t, db1.Create(event1).Error)
|
|
require.NoError(t, db2.Create(event2).Error)
|
|
|
|
// Verify isolation: each DB only has its own events
|
|
var count1 int64
|
|
|
|
db1.Model(&database.Event{}).Count(&count1)
|
|
assert.Equal(t, int64(1), count1)
|
|
|
|
var count2 int64
|
|
|
|
db2.Model(&database.Event{}).Count(&count2)
|
|
assert.Equal(t, int64(1), count2)
|
|
|
|
// Delete webhook1's DB, webhook2 should be unaffected
|
|
require.NoError(t, mgr.DeleteDB(webhook1))
|
|
assert.False(t, mgr.DBExists(webhook1))
|
|
assert.True(t, mgr.DBExists(webhook2))
|
|
|
|
// webhook2's data should still be accessible
|
|
var events []database.Event
|
|
|
|
require.NoError(t, db2.Find(&events).Error)
|
|
|
|
// require, not assert: this is exactly the regression the test
|
|
// guards, so the empty slice is the expected failure, and a
|
|
// non-fatal length check would index into it on the next line and
|
|
// panic the whole package test binary instead of failing here.
|
|
require.Len(t, events, 1)
|
|
assert.Equal(t, "PUT", events[0].Method)
|
|
}
|
|
|
|
func TestWebhookDBManager_CloseAll(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
mgr, lc := setupTestWebhookDBManager(t)
|
|
ctx := context.Background()
|
|
require.NoError(t, lc.Start(ctx))
|
|
|
|
// Create a few DBs
|
|
for range 3 {
|
|
require.NoError(
|
|
t,
|
|
mgr.CreateDB(uuid.New().String()),
|
|
)
|
|
}
|
|
|
|
// CloseAll should close all connections without error
|
|
require.NoError(t, mgr.CloseAll())
|
|
|
|
// Stop lifecycle (CloseAll already called)
|
|
require.NoError(t, lc.Stop(ctx))
|
|
}
|