Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8328016bec |
@@ -1700,6 +1700,22 @@ retries) is individually logged for full observability.
|
|||||||
|
|
||||||
**Relations:** Belongs to Delivery.
|
**Relations:** Belongs to Delivery.
|
||||||
|
|
||||||
|
#### Event-tier indexes
|
||||||
|
|
||||||
|
Beyond the primary keys, the per-webhook event databases carry secondary
|
||||||
|
indexes on the columns the background work reads by, each created by
|
||||||
|
`AutoMigrate` on a fresh and on an existing database:
|
||||||
|
|
||||||
|
| Column | Serves |
|
||||||
|
| ------------------------------ | ------ |
|
||||||
|
| `deliveries.status` | The recovery and sweep queries that select deliveries by status once a minute |
|
||||||
|
| `deliveries.event_id` | Loading a page of the event log, which reads deliveries by event |
|
||||||
|
| `delivery_results.delivery_id` | Loading a page of the event log, which reads results by delivery |
|
||||||
|
| `events.created_at` | Retention, which deletes events by age |
|
||||||
|
|
||||||
|
The `events.resubmitted_from_id` column is also indexed, to resolve the
|
||||||
|
resubmit relationship both ways in the event log.
|
||||||
|
|
||||||
#### Common Fields
|
#### Common Fields
|
||||||
|
|
||||||
Every entity except `Setting` includes these fields from `BaseModel`.
|
Every entity except `Setting` includes these fields from `BaseModel`.
|
||||||
|
|||||||
@@ -17,12 +17,12 @@ import (
|
|||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
"sneak.berlin/go/webhooker/internal/banner"
|
"sneak.berlin/go/webhooker/internal/banner"
|
||||||
"sneak.berlin/go/webhooker/internal/config"
|
"sneak.berlin/go/webhooker/internal/config"
|
||||||
"sneak.berlin/go/webhooker/internal/datadir"
|
|
||||||
"sneak.berlin/go/webhooker/internal/gormlog"
|
"sneak.berlin/go/webhooker/internal/gormlog"
|
||||||
"sneak.berlin/go/webhooker/internal/logger"
|
"sneak.berlin/go/webhooker/internal/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
dataDirPerm = 0750
|
||||||
randomPasswordLen = 16
|
randomPasswordLen = 16
|
||||||
sessionKeyLen = 32
|
sessionKeyLen = 32
|
||||||
)
|
)
|
||||||
@@ -185,9 +185,7 @@ func (d *Database) connect() error {
|
|||||||
// caller's decision.
|
// caller's decision.
|
||||||
func (d *Database) connectTo(dataDir string) error {
|
func (d *Database) connectTo(dataDir string) error {
|
||||||
// Ensure the data directory exists before opening the database.
|
// Ensure the data directory exists before opening the database.
|
||||||
// datadir.DirPerm is the single source of the directory mode; this
|
err := os.MkdirAll(dataDir, dataDirPerm)
|
||||||
// package creates the directory too, since either may run first.
|
|
||||||
err := os.MkdirAll(dataDir, datadir.DirPerm)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf(
|
return fmt.Errorf(
|
||||||
"creating data directory %s: %w",
|
"creating data directory %s: %w",
|
||||||
|
|||||||
@@ -0,0 +1,72 @@
|
|||||||
|
package database_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/google/uuid"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
|
)
|
||||||
|
|
||||||
|
// indexedColumn names a secondary index by the model and struct field
|
||||||
|
// GORM derives the index name from.
|
||||||
|
type indexedColumn struct {
|
||||||
|
model any
|
||||||
|
field string
|
||||||
|
}
|
||||||
|
|
||||||
|
// eventTierIndexes are the columns the background work reads by: the
|
||||||
|
// recovery and sweep queries (status), the event log (event_id and
|
||||||
|
// delivery_id) and retention (created_at).
|
||||||
|
var eventTierIndexes = []indexedColumn{
|
||||||
|
{&database.Delivery{}, "Status"},
|
||||||
|
{&database.Delivery{}, "EventID"},
|
||||||
|
{&database.DeliveryResult{}, "DeliveryID"},
|
||||||
|
{&database.Event{}, "CreatedAt"},
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestWebhookDBManager_OpenAddsEventTierIndexes verifies that opening a
|
||||||
|
// per-webhook database that predates these indexes creates them, so the
|
||||||
|
// queries above stop scanning whole tables. It stands in for an older
|
||||||
|
// database file by dropping the indexes AutoMigrate just created, then
|
||||||
|
// reopening the same file.
|
||||||
|
func TestWebhookDBManager_OpenAddsEventTierIndexes(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, err := mgr.GetDB(webhookID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
// A fresh database has them.
|
||||||
|
for _, ix := range eventTierIndexes {
|
||||||
|
require.True(t, db.Migrator().HasIndex(ix.model, ix.field))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stand in for a database file created before the indexes existed.
|
||||||
|
for _, ix := range eventTierIndexes {
|
||||||
|
require.NoError(t, db.Migrator().DropIndex(ix.model, ix.field))
|
||||||
|
require.False(t, db.Migrator().HasIndex(ix.model, ix.field))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Drop the cached connection so the next open reopens the file and
|
||||||
|
// runs AutoMigrate against it, as a restart would.
|
||||||
|
require.NoError(t, mgr.CloseAll())
|
||||||
|
|
||||||
|
db, err = mgr.GetDB(webhookID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
for _, ix := range eventTierIndexes {
|
||||||
|
assert.True(t, db.Migrator().HasIndex(ix.model, ix.field),
|
||||||
|
"opening the existing database should create the index on %s",
|
||||||
|
ix.field)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -32,9 +32,9 @@ func (s DeliveryStatus) Terminal() bool {
|
|||||||
type Delivery struct {
|
type Delivery struct {
|
||||||
BaseModel
|
BaseModel
|
||||||
|
|
||||||
EventID string `gorm:"type:uuid;not null" json:"eventId"`
|
EventID string `gorm:"type:uuid;not null;index" json:"eventId"`
|
||||||
TargetID string `gorm:"type:uuid;not null" json:"targetId"`
|
TargetID string `gorm:"type:uuid;not null" json:"targetId"`
|
||||||
Status DeliveryStatus `gorm:"not null;default:'pending'" json:"status"`
|
Status DeliveryStatus `gorm:"not null;default:'pending';index" json:"status"`
|
||||||
|
|
||||||
// Relations
|
// Relations
|
||||||
Event Event `json:"event,omitzero"`
|
Event Event `json:"event,omitzero"`
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ package database
|
|||||||
type DeliveryResult struct {
|
type DeliveryResult struct {
|
||||||
BaseModel
|
BaseModel
|
||||||
|
|
||||||
DeliveryID string `gorm:"type:uuid;not null" json:"deliveryId"`
|
DeliveryID string `gorm:"type:uuid;not null;index" json:"deliveryId"`
|
||||||
AttemptNum int `gorm:"not null" json:"attemptNum"`
|
AttemptNum int `gorm:"not null" json:"attemptNum"`
|
||||||
Success bool `json:"success"`
|
Success bool `json:"success"`
|
||||||
StatusCode int `json:"statusCode,omitempty"`
|
StatusCode int `json:"statusCode,omitempty"`
|
||||||
|
|||||||
@@ -1,9 +1,17 @@
|
|||||||
package database
|
package database
|
||||||
|
|
||||||
|
import "time"
|
||||||
|
|
||||||
// Event represents a captured webhook event
|
// Event represents a captured webhook event
|
||||||
type Event struct {
|
type Event struct {
|
||||||
BaseModel
|
BaseModel
|
||||||
|
|
||||||
|
// CreatedAt overrides BaseModel.CreatedAt only to add an index:
|
||||||
|
// retention deletes events by age, so events.created_at is queried
|
||||||
|
// on every sweep. The other tables keep the unindexed BaseModel
|
||||||
|
// field.
|
||||||
|
CreatedAt time.Time `gorm:"index" json:"createdAt"`
|
||||||
|
|
||||||
WebhookID string `gorm:"type:uuid;not null" json:"webhookId"`
|
WebhookID string `gorm:"type:uuid;not null" json:"webhookId"`
|
||||||
EntrypointID string `gorm:"type:uuid;not null" json:"entrypointId"`
|
EntrypointID string `gorm:"type:uuid;not null" json:"entrypointId"`
|
||||||
|
|
||||||
|
|||||||
@@ -13,7 +13,6 @@ import (
|
|||||||
"gorm.io/driver/sqlite"
|
"gorm.io/driver/sqlite"
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
"sneak.berlin/go/webhooker/internal/config"
|
"sneak.berlin/go/webhooker/internal/config"
|
||||||
"sneak.berlin/go/webhooker/internal/datadir"
|
|
||||||
"sneak.berlin/go/webhooker/internal/gormlog"
|
"sneak.berlin/go/webhooker/internal/gormlog"
|
||||||
"sneak.berlin/go/webhooker/internal/logger"
|
"sneak.berlin/go/webhooker/internal/logger"
|
||||||
)
|
)
|
||||||
@@ -54,9 +53,8 @@ func NewWebhookDBManager(
|
|||||||
log: params.Logger.Get(),
|
log: params.Logger.Get(),
|
||||||
}
|
}
|
||||||
|
|
||||||
// Create data directory if it doesn't exist. datadir.DirPerm is the
|
// Create data directory if it doesn't exist
|
||||||
// single source of the directory mode; either package may run first.
|
err := os.MkdirAll(m.dataDir, dataDirPerm)
|
||||||
err := os.MkdirAll(m.dataDir, datadir.DirPerm)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf(
|
return nil, fmt.Errorf(
|
||||||
"creating data directory %s: %w",
|
"creating data directory %s: %w",
|
||||||
|
|||||||
@@ -29,11 +29,9 @@ import (
|
|||||||
// process that was killed with SIGKILL blocks nothing.
|
// process that was killed with SIGKILL blocks nothing.
|
||||||
const LockFileName = "webhooker.lock"
|
const LockFileName = "webhooker.lock"
|
||||||
|
|
||||||
// DirPerm is the mode DATA_DIR is created with. It is the single
|
// dirPerm is the mode Acquire creates DATA_DIR with. It matches what
|
||||||
// source of that mode: internal/database consumes it rather than
|
// internal/database uses, since whichever runs first creates it.
|
||||||
// keeping its own copy, so the two packages that both create the
|
const dirPerm = 0o750
|
||||||
// directory cannot drift into disagreeing about its permissions.
|
|
||||||
const DirPerm = 0o750
|
|
||||||
|
|
||||||
// ErrLocked reports that another live process holds the data
|
// ErrLocked reports that another live process holds the data
|
||||||
// directory. Callers that need to know whether a deployment is running
|
// directory. Callers that need to know whether a deployment is running
|
||||||
@@ -66,7 +64,7 @@ func Acquire(dir string) (*Lock, error) {
|
|||||||
return nil, ErrNoDir
|
return nil, ErrNoDir
|
||||||
}
|
}
|
||||||
|
|
||||||
err := os.MkdirAll(dir, DirPerm)
|
err := os.MkdirAll(dir, dirPerm)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf(
|
return nil, fmt.Errorf(
|
||||||
"creating data directory %s: %w", dir, err,
|
"creating data directory %s: %w", dir, err,
|
||||||
|
|||||||
Reference in New Issue
Block a user