6 Commits

Author SHA1 Message Date
d61d9dc1c1 Drop the stale open-work claim from the TODO status section
All checks were successful
check / check (push) Successful in 2m56s
The milestone is the authoritative count and the section already says
so; the lead-in asserted work remaining independently of it.
2026-08-24 15:52:44 +00:00
b0a011f6b4 Render unknown for a zero CreatedAt in Slack/Mattermost messages (closes #298)
Some checks failed
check / check (push) Superseded by a newer commit; never tested
2026-08-24 17:52:25 +02:00
5976a4a98f Carry the event's receipt time into every delivery (closes #257)
All checks were successful
check / check (push) Successful in 3m4s
2026-08-24 06:44:24 +02:00
b2c9acdaa6 Correct four documentation claims ahead of the 1.0.0 tag
All checks were successful
check / check (push) Successful in 7s
2026-08-24 06:25:08 +02:00
af3703d748 Close the two remaining delivery terminal-state gaps (closes #107)
All checks were successful
check / check (push) Successful in 3m16s
2026-08-24 05:12:02 +02:00
322d9a6d6b Create every SQLite file 0600 (closes #255)
All checks were successful
check / check (push) Successful in 3m17s
2026-08-24 04:33:40 +02:00
12 changed files with 1728 additions and 43 deletions

View File

@@ -686,6 +686,44 @@ databases written by `database` targets (`archive-{uuid}.db`). Mount
this as a persistent volume to preserve data across container this as a persistent volume to preserve data across container
restarts. restarts.
**The bind-mounted directory must be owned by UID 1000, or the
container does not start.** Docker creates a `-v` source path that
does not exist yet as `root:root`, and the process runs as UID 1000,
so it cannot take its `DATA_DIR` lock:
```
webhooker: locking data directory /var/lib/webhooker: open
/var/lib/webhooker/webhooker.lock: permission denied
```
It exits non-zero at that point, before opening any database. Create
the directory ahead of the first `docker run`:
```bash
mkdir -p /path/to/data
chown 1000:1000 /path/to/data
chmod 750 /path/to/data
```
The same `chown` is what a restore needs — see step 4 of
[Restore](#restore). A **named volume** does not have this problem:
Docker copies the image's ownership onto a volume it initializes, and
the image creates `/var/lib/webhooker` owned by `webhooker`.
**The file modes are not yours to set, and do not depend on the
directory.** `webhooker.db` holds target configuration in plaintext —
bearer tokens, API keys, Slack webhook URLs — along with the session
encryption key, so webhooker creates every SQLite file it owns `0600`:
each database and both of its `-wal` and `-shm` sidecars, across all
three tiers. Files an earlier build left `0644` are tightened when
they are opened. A `DATA_DIR` webhooker creates itself is `0750`, but
a bind mount supplies its own directory and Docker's default for one
it creates is `0755`; the `0600` files hold there regardless. The
`chmod 750` above is defence in depth — it stops other local users
listing the directory and learning your webhook UUIDs from the
`events-{uuid}.db` filenames — not the barrier protecting the
credentials.
## Deployment behind a reverse proxy ## Deployment behind a reverse proxy
webhooker terminates no TLS of its own. It serves plaintext HTTP and webhooker terminates no TLS of its own. It serves plaintext HTTP and
@@ -1196,6 +1234,16 @@ webhooker solves this by acting as a durable intermediary:
backoff. Every delivery attempt is logged with status codes, response backoff. Every delivery attempt is logged with status codes, response
bodies, and timing. bodies, and timing.
**That guarantee is at-least-once, not exactly-once.** When a send
reaches its target but the write recording that outcome fails, the
delivery is deliberately left in a recoverable state rather than
marked done — losing a delivery is the worse failure — so the
pending sweep picks it up about fifteen minutes later, or the next
restart does, and the target receives a payload it already got.
webhooker adds no delivery identifier of its own to an outbound
request, so **make your receiver idempotent** against whatever the
payload itself carries.
3. **Observability** — Full request/response logging for every webhook 3. **Observability** — Full request/response logging for every webhook
received and every delivery attempted. Prometheus metrics expose received and every delivery attempted. Prometheus metrics expose
volume, latency, and error rates. The web UI provides real-time volume, latency, and error rates. The web UI provides real-time
@@ -1642,6 +1690,11 @@ webhooker uses **separate SQLite database files**: a main application
database for configuration data and per-webhook databases for event database for configuration data and per-webhook databases for event
storage. All database files live in the `DATA_DIR` directory. storage. All database files live in the `DATA_DIR` directory.
Every one of them is created `0600`, and so is each `-wal` and `-shm`
sidecar. See
[Running with Docker](#running-with-docker) for what that does and
does not protect.
**Main Application Database** (`{DATA_DIR}/webhooker.db`) — stores **Main Application Database** (`{DATA_DIR}/webhooker.db`) — stores
configuration and application state: configuration and application state:
@@ -2573,12 +2626,15 @@ abuse limit later; they are tracked as future work.
| `POST` | `/source/{id}/edit` | Edit webhook submission | | `POST` | `/source/{id}/edit` | Edit webhook submission |
| `POST` | `/source/{id}/delete` | Delete webhook | | `POST` | `/source/{id}/delete` | Delete webhook |
| `GET` | `/source/{id}/logs` | Webhook event logs | | `GET` | `/source/{id}/logs` | Webhook event logs |
| `GET` | `/source/{id}/logs/{eventID}/body` | Download an event's full stored body. The log page renders each body only up to its cap, so this is the only route that serves a whole one; it is offered wherever a body is shown truncated |
| `POST` | `/source/{id}/deliveries/{deliveryID}/replay` | Replay a finished delivery: creates a new delivery for the same event against the target's current configuration (30 per minute per bucket, then `429`) | | `POST` | `/source/{id}/deliveries/{deliveryID}/replay` | Replay a finished delivery: creates a new delivery for the same event against the target's current configuration (30 per minute per bucket, then `429`) |
| `POST` | `/source/{id}/events/{eventID}/resubmit` | Resubmit a stored event: creates a new event copying it and fans that out to every currently active target (30 per minute per bucket, then `429`) | | `POST` | `/source/{id}/events/{eventID}/resubmit` | Resubmit a stored event: creates a new event copying it and fans that out to every currently active target (30 per minute per bucket, then `429`) |
| `POST` | `/source/{id}/entrypoints` | Add entrypoint to webhook | | `POST` | `/source/{id}/entrypoints` | Add entrypoint to webhook |
| `POST` | `/source/{id}/entrypoints/{entrypointID}/delete` | Delete an entrypoint | | `POST` | `/source/{id}/entrypoints/{entrypointID}/delete` | Delete an entrypoint |
| `POST` | `/source/{id}/entrypoints/{entrypointID}/toggle` | Enable or disable an entrypoint | | `POST` | `/source/{id}/entrypoints/{entrypointID}/toggle` | Enable or disable an entrypoint |
| `POST` | `/source/{id}/targets` | Add target to webhook | | `POST` | `/source/{id}/targets` | Add target to webhook |
| `GET` | `/source/{id}/targets/{targetID}/edit` | Edit target form. The one page that renders a target's destination URL and header values in full, rather than masked |
| `POST` | `/source/{id}/targets/{targetID}/edit` | Edit target submission |
| `POST` | `/source/{id}/targets/{targetID}/delete` | Delete a target | | `POST` | `/source/{id}/targets/{targetID}/delete` | Delete a target |
| `POST` | `/source/{id}/targets/{targetID}/toggle` | Enable or disable a target | | `POST` | `/source/{id}/targets/{targetID}/toggle` | Enable or disable a target |
@@ -2617,6 +2673,8 @@ webhooker/
├── internal/ ├── internal/
│ ├── banner/ │ ├── banner/
│ │ └── banner.go # Ruled block for the one credential shown in the clear │ │ └── banner.go # Ruled block for the one credential shown in the clear
│ ├── ciscript/
│ │ └── doc.go # Tests for the CI shell scripts in script/; no runtime code
│ ├── resetpw/ │ ├── resetpw/
│ │ └── resetpw.go # `webhooker resetpw`: set an account's password, stopped deployments only │ │ └── resetpw.go # `webhooker resetpw`: set an account's password, stopped deployments only
│ ├── config/ │ ├── config/
@@ -2686,13 +2744,17 @@ webhooker/
│ │ ├── ratelimit.go # Per-IP rate limiting middleware (go-chi/httprate) │ │ ├── ratelimit.go # Per-IP rate limiting middleware (go-chi/httprate)
│ │ ├── loginguard.go # Login failure counters and the Argon2id verification semaphore │ │ ├── loginguard.go # Login failure counters and the Argon2id verification semaphore
│ │ └── testing.go # NewForTest: Middleware without the fx lifecycle │ │ └── testing.go # NewForTest: Middleware without the fx lifecycle
│ ├── reqtls/
│ │ └── reqtls.go # IsTLS: the one TLS predicate, r.TLS or X-Forwarded-Proto
│ ├── server/ │ ├── server/
│ │ ├── server.go # Server struct, fx lifecycle, signal handling │ │ ├── server.go # Server struct, fx lifecycle, signal handling
│ │ ├── http.go # HTTP server setup with timeouts │ │ ├── http.go # HTTP server setup with timeouts
│ │ └── routes.go # All route definitions │ │ └── routes.go # All route definitions
── session/ ── session/
├── session.go # Cookie-based session management ├── session.go # Cookie-based session management
└── testing.go # NewForTest: Session without the fx lifecycle └── testing.go # NewForTest: Session without the fx lifecycle
│ └── versionscript/
│ └── doc.go # Tests for script/version and the build files that use it
├── static/ ├── static/
│ ├── static.go # //go:embed directive │ ├── static.go # //go:embed directive
│ ├── css/input.css # Tailwind input, source for tailwind.css (make css) │ ├── css/input.css # Tailwind input, source for tailwind.css (make css)

41
TODO.md
View File

@@ -18,18 +18,27 @@ Issue branches do NOT touch this file — the manager maintains it on
# Status # Status
1.0.0 is open, with work remaining. The milestone The milestone (https://git.eeqj.de/sneak/webhooker/milestone/9) is the
(https://git.eeqj.de/sneak/webhooker/milestone/9) is the authoritative authoritative list, and the only place to read a count or a state of
list, and the only place to read a count or a state of play from. This play from. This file records where the project is, not what is in
file records where the project is, not what is in flight: a sentence flight: a sentence whose truth depends on a branch being unmerged is
whose truth depends on a branch being unmerged is wrong the moment it wrong the moment it merges, and this file has been wrong that way
merges, and this file has been wrong that way before. before.
The tag is held on a durability defect The durability defect that held the tag has landed
(https://git.eeqj.de/sneak/webhooker/issues/256): a concurrent reader (https://git.eeqj.de/sneak/webhooker/issues/256, commit `8d64259`).
of a per-webhook event database strands delivered webhooks at Every SQLite handle opens with WAL journaling and a busy timeout, a
`pending`, and the next restart re-delivers them. That issue gates bookkeeping write that fails leaves its delivery in a recoverable
`v1.0.0`, and is where the fix's own state is tracked. state rather than a lying one, and recovery skips a delivery that
already has a successful result row. Final pre-tag verification
exercised it and confirmed it holds. Whatever the milestone still
shows open is what remains before `v1.0.0`.
Delivery is at-least-once by design, not by accident: a send whose
result row does not land is attempted again, so a receiver can see a
duplicate. That is deliberate — the alternative is a silent lost
delivery — and the README says so under Rationale. It is not a defect
to re-file.
One caveat on reading a green check: a docs-only commit deliberately One caveat on reading a green check: a docs-only commit deliberately
replays from the layer cache replays from the layer cache
@@ -39,11 +48,11 @@ commit invalidates the `COPY` layer and genuinely executes.
# Next Step # Next Step
Land https://git.eeqj.de/sneak/webhooker/issues/256, then clear the Clear the rest of the open 1.0.0 milestone
rest of the open 1.0.0 milestone and tag `v1.0.0`. Merging `next` into (https://git.eeqj.de/sneak/webhooker/milestone/9) and tag `v1.0.0`.
`main` is a separate act from tagging and waits on neither of those: Merging `next` into `main` is a separate act from tagging and waits on
`next` is kept mergeable at all times, which is the point of the neither of those: `next` is kept mergeable at all times, which is the
branch. point of the branch.
# Completed Steps # Completed Steps

View File

@@ -0,0 +1,240 @@
package database_test
import (
"context"
"io/fs"
"net/http"
"os"
"path/filepath"
"testing"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/fx/fxtest"
"sneak.berlin/go/webhooker/internal/config"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/globals"
"sneak.berlin/go/webhooker/internal/logger"
)
// ownerOnly is the mode every SQLite file the service owns must have.
// Spelled out rather than referencing database.SQLiteFilePerm so the
// test fails if the constant itself is loosened.
const ownerOnly fs.FileMode = 0o600
// requireOwnerOnly asserts that path exists and is readable and
// writable by its owner and by nobody else.
func requireOwnerOnly(t *testing.T, path string) {
t.Helper()
info, err := os.Stat(path)
require.NoError(t, err, "%s must exist", path)
assert.Equal(
t,
ownerOnly,
info.Mode().Perm(),
"%s holds credentials and must not be readable by "+
"anyone but its owner",
path,
)
}
// requireDatabaseSetOwnerOnly asserts the mode of a database file and
// of both WAL sidecars. The sidecars carry the same rows as the
// database, so tightening only the main file fixes nothing.
func requireDatabaseSetOwnerOnly(t *testing.T, dbPath string) {
t.Helper()
requireOwnerOnly(t, dbPath)
requireOwnerOnly(t, dbPath+"-wal")
requireOwnerOnly(t, dbPath+"-shm")
}
// TestMainDatabaseFilesAreOwnerOnly covers the tier the defect was
// reported against: webhooker.db holds targets.config in plaintext —
// bearer tokens, API keys, Slack webhook URLs — and the session
// encryption key.
func TestMainDatabaseFilesAreOwnerOnly(t *testing.T) {
t.Parallel()
lc := fxtest.NewLifecycle(t)
l, err := logger.New(lc, logger.LoggerParams{
Globals: &globals.Globals{
Appname: testAppname,
Version: testVersion,
},
})
require.NoError(t, err)
// A directory the application creates itself, not one t.TempDir
// made at 0700, so the mode below is the application's.
dataDir := filepath.Join(t.TempDir(), "data")
db, err := database.New(lc, database.DatabaseParams{
Config: &config.Config{DataDir: dataDir},
Logger: l,
})
require.NoError(t, err)
ctx := context.Background()
require.NoError(t, lc.Start(ctx))
defer func() { require.NoError(t, lc.Stop(ctx)) }()
// Write through the real model so the WAL is populated and both
// sidecars are on disk while the handle is open.
require.NoError(t, db.DB().Create(&database.Webhook{
Name: testWebhookName,
}).Error)
requireDatabaseSetOwnerOnly(
t, filepath.Join(dataDir, database.MainDBFileName),
)
// The data directory grants nothing to `other`. Asserted as a
// property rather than as an exact 0750, because MkdirAll applies
// the ambient umask: the exact mode is the developer's umask as
// much as the application's request, and pinning it would make
// `make check` pass or fail on where it is run. The group bits are
// deliberately left unasserted — deployments may rely on them.
info, err := os.Stat(dataDir)
require.NoError(t, err)
assert.Zero(
t,
info.Mode().Perm()&0o007,
"the data directory must not be world-accessible",
)
}
// TestPerWebhookEventDatabaseFilesAreOwnerOnly covers the events-*.db
// tier. These carry no credential canaries since
// https://git.eeqj.de/sneak/webhooker/issues/206, but they hold every
// received request body and header.
func TestPerWebhookEventDatabaseFilesAreOwnerOnly(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)
require.NoError(t, db.Create(&database.Event{
WebhookID: webhookID,
EntrypointID: uuid.New().String(),
Method: http.MethodPost,
Body: "{}",
}).Error)
requireDatabaseSetOwnerOnly(t, mgr.DBPath(webhookID))
}
// TestArchiveDatabaseFilesAreOwnerOnly covers the archive-*.db tier.
// internal/delivery builds that path and opens it through OpenSQLite,
// the same single open path exercised here, so the mode is settled for
// all three tiers in one place.
func TestArchiveDatabaseFilesAreOwnerOnly(t *testing.T) {
t.Parallel()
ctx := context.Background()
path := filepath.Join(
t.TempDir(), "archive-"+uuid.New().String()+".db",
)
sqlDB, err := database.OpenSQLite(path, database.SQLiteModeCreate)
require.NoError(t, err)
defer func() { require.NoError(t, sqlDB.Close()) }()
_, err = sqlDB.ExecContext(ctx, "create table t (id integer)")
require.NoError(t, err)
requireDatabaseSetOwnerOnly(t, path)
}
// TestOpenSQLiteTightensFilesLeftWorldReadable is the upgrade case: a
// data directory an earlier build left at 0644, including a
// developer's own scratch directory, is fixed when it is opened rather
// than staying exposed until it is recreated.
func TestOpenSQLiteTightensFilesLeftWorldReadable(t *testing.T) {
t.Parallel()
dir := t.TempDir()
path := filepath.Join(dir, database.MainDBFileName)
// A database and both sidecars as the pre-fix build left them.
for _, p := range []string{path, path + "-wal", path + "-shm"} {
require.NoError(t, os.WriteFile(p, nil, 0o644)) //nolint:gosec // the mode under test
}
sqlDB, err := database.OpenSQLite(path, database.SQLiteModeCreate)
require.NoError(t, err)
require.NoError(t, sqlDB.Close())
requireDatabaseSetOwnerOnly(t, path)
}
// TestOpenSQLiteExistingModeDoesNotCreateTheFile guards the mechanism
// the fix uses: OpenSQLite now creates the database file itself, and
// must not do so for a caller that asked for an existing database. An
// empty file materialized here would turn a missing-database error
// into a silently empty one.
func TestOpenSQLiteExistingModeDoesNotCreateTheFile(t *testing.T) {
t.Parallel()
ctx := context.Background()
path := filepath.Join(t.TempDir(), "absent.db")
sqlDB, err := database.OpenSQLite(path, database.SQLiteModeExisting)
if err == nil {
// sql.Open is lazy: force the connection that fails.
require.Error(t, sqlDB.PingContext(ctx))
require.NoError(t, sqlDB.Close())
}
_, statErr := os.Stat(path)
assert.ErrorIs(t, statErr, fs.ErrNotExist)
}
// TestReopenAfterRestartKeepsFilesOwnerOnly is the restart case: a
// process that closed its files must be able to open them again at
// 0600, including through a gorm handle, and the sidecars must come
// back at 0600 too rather than at SQLite's own default.
func TestReopenAfterRestartKeepsFilesOwnerOnly(t *testing.T) {
t.Parallel()
ctx := context.Background()
dir := t.TempDir()
path := filepath.Join(dir, database.MainDBFileName)
first, err := database.OpenSQLite(path, database.SQLiteModeCreate)
require.NoError(t, err)
_, err = first.ExecContext(ctx, "create table t (id integer)")
require.NoError(t, err)
require.NoError(t, first.Close())
second, err := database.OpenSQLite(path, database.SQLiteModeCreate)
require.NoError(t, err)
defer func() { require.NoError(t, second.Close()) }()
_, err = second.ExecContext(ctx, "insert into t (id) values (1)")
require.NoError(t, err)
requireDatabaseSetOwnerOnly(t, path)
var got int
require.NoError(t,
second.QueryRowContext(ctx, "select id from t").Scan(&got))
assert.Equal(t, 1, got)
}

View File

@@ -2,8 +2,11 @@ package database
import ( import (
"database/sql" "database/sql"
"errors"
"fmt" "fmt"
"io/fs"
"net/url" "net/url"
"os"
"time" "time"
_ "modernc.org/sqlite" // Pure Go SQLite driver _ "modernc.org/sqlite" // Pure Go SQLite driver
@@ -72,6 +75,90 @@ const (
sqliteConnMaxIdleTime = time.Minute sqliteConnMaxIdleTime = time.Minute
) )
// SQLiteFilePerm is the mode every SQLite file this service owns is
// created with and held at: owner read/write, nothing for group or
// other.
//
// These files hold credentials in plaintext. The main database stores
// `targets.config` — bearer tokens, API keys, Slack webhook URLs — and
// the session encryption key. SQLite left to itself creates them 0644
// (see reserveSQLiteFile), which made the 0750 data directory the only
// barrier; a bind-mounted directory supplied at 0755 removes it and
// every local user on the host can read every stored credential.
//
// This is a file-mode fix and not encryption at rest. An unattended
// process needs a key it can read without a human, so the key lands
// beside the data and an attacker who can read the database can read
// it too. See https://git.eeqj.de/sneak/webhooker/issues/212.
const SQLiteFilePerm fs.FileMode = 0o600
// reserveSQLiteFile puts path at SQLiteFilePerm before the driver ever
// touches it, and tightens any sidecar already on disk.
//
// The mode has to be settled here rather than by a chmod after opening,
// because SQLite picks it: robust_open substitutes
// SQLITE_DEFAULT_FILE_PERMISSIONS (0644) whenever it is handed mode 0,
// and findCreateFileMode yields 0 for a main database opened by URI
// with no `modeof` parameter. A chmod afterwards would leave a window
// in which the credentials are on disk world-readable.
//
// Creating the file ourselves also settles the sidecars, which is the
// half that could quietly not work. SQLite does not create those at a
// mode we choose — it derives both from the main database file:
// `-wal` through findCreateFileMode, which stats the path with the
// suffix stripped, and `-shm` in unixOpenSharedMemory from an fstat of
// the already-open database descriptor. A main file at 0600 therefore
// produces sidecars at 0600. A zero-length file is a valid empty
// database, so reserving it changes nothing else.
//
// create says whether the caller is opening in a mode that may create
// the database. When it is false a missing file is left missing, so
// SQLite still reports the absence rather than this function
// materializing an empty database the caller asked not to create.
//
// Chmod of a file that already exists is what tightens a data
// directory an earlier build left at 0644 — including a developer's
// own scratch directory — without any migration machinery.
func reserveSQLiteFile(path string, create bool) error {
if create {
// gosec G304: the path is the database file the caller asked
// to open, and the driver is about to open the same path
// anyway. Creating it here is what fixes its mode.
f, err := os.OpenFile( //nolint:gosec // see above
path, os.O_RDWR|os.O_CREATE, SQLiteFilePerm,
)
if err != nil {
return fmt.Errorf("creating %s: %w", path, err)
}
err = f.Close()
if err != nil {
return fmt.Errorf("closing %s: %w", path, err)
}
}
// O_CREATE leaves an existing file's mode alone, and umask can only
// have narrowed a new one. Chmod settles both cases at exactly
// SQLiteFilePerm.
for _, p := range append(
[]string{path}, sqliteSidecarPaths(path)...,
) {
err := os.Chmod(p, SQLiteFilePerm)
if err != nil && !errors.Is(err, fs.ErrNotExist) {
return fmt.Errorf("securing %s: %w", p, err)
}
}
return nil
}
// sqliteSidecarPaths returns the files SQLite maintains beside a
// database under WAL. They carry the same rows as the database itself,
// so a fix that tightens only the main file has fixed nothing.
func sqliteSidecarPaths(path string) []string {
return []string{path + "-wal", path + "-shm"}
}
// SQLiteDSN builds the connection string for one database file. // SQLiteDSN builds the connection string for one database file.
// //
// mode is the SQLite URI open mode: "rwc" to create the file when it // mode is the SQLite URI open mode: "rwc" to create the file when it
@@ -138,9 +225,17 @@ func SQLiteDSN(path, mode string) string {
// durability settings and pool bounds applied. mode is the SQLite URI // durability settings and pool bounds applied. mode is the SQLite URI
// open mode ("rwc" or "rw"). // open mode ("rwc" or "rw").
// //
// The file and its WAL sidecars are settled at SQLiteFilePerm before
// the driver sees the path; see reserveSQLiteFile.
//
// The handle is returned rather than a *gorm.DB because the callers // The handle is returned rather than a *gorm.DB because the callers
// wrap it in gorm themselves with their own logger. // wrap it in gorm themselves with their own logger.
func OpenSQLite(path, mode string) (*sql.DB, error) { func OpenSQLite(path, mode string) (*sql.DB, error) {
err := reserveSQLiteFile(path, mode == SQLiteModeCreate)
if err != nil {
return nil, err
}
sqlDB, err := sql.Open("sqlite", SQLiteDSN(path, mode)) sqlDB, err := sql.Open("sqlite", SQLiteDSN(path, mode))
if err != nil { if err != nil {
return nil, fmt.Errorf( return nil, fmt.Errorf(

View File

@@ -4,6 +4,7 @@ package delivery
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"log/slog" "log/slog"
"net/http" "net/http"
@@ -439,7 +440,7 @@ func (e *Engine) processNewTask(
event := buildEventFromTask(task) event := buildEventFromTask(task)
event, err = e.resolveEventBody( event, err = e.hydrateEvent(
webhookDB, event, task, webhookDB, event, task,
) )
if err != nil { if err != nil {
@@ -505,9 +506,13 @@ func (e *Engine) processRetryTask(
return return
} }
if e.abandonRetryForMissingTarget(webhookDB, d, task) {
return
}
event := buildEventFromTask(task) event := buildEventFromTask(task)
event, err = e.resolveEventBody( event, err = e.hydrateEvent(
webhookDB, event, task, webhookDB, event, task,
) )
if err != nil { if err != nil {
@@ -529,6 +534,64 @@ func (e *Engine) processRetryTask(
e.processDelivery(ctx, webhookDB, d, task) e.processDelivery(ctx, webhookDB, d, task)
} }
// abandonRetryForMissingTarget stops a retry chain whose target has
// been deleted, and reports whether it did.
//
// A scheduled retry lives in memory as a time.AfterFunc holding the
// target's configuration as it was when the chain began, and nothing
// else on this path reads the target row. Without this check a
// deletion stops nothing: the timer keeps firing and keeps sending to
// the destination the operator removed, for the whole remaining
// backoff chain. Terminalising in the recovery and sweep paths alone
// is not enough, because those only see the delivery once nothing
// holds it in memory — which is to say after a restart.
//
// The worker already owns this delivery, so the terminal write happens
// here directly, exactly as a target's own Deliver fails one. Claiming
// it again through the recovery gate would only fail against the
// reference the worker itself is holding.
//
// A lookup that fails for any other reason is not a deletion — it is
// the main database being unreadable — and the delivery goes ahead as
// it did before. A guard that terminally failed deliveries on a
// transient fault would be worse than the bug it fixes.
func (e *Engine) abandonRetryForMissingTarget(
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
) bool {
_, err := e.loadTarget(task.TargetID)
if err == nil {
return false
}
if !errors.Is(err, gorm.ErrRecordNotFound) {
e.log.Warn(
"could not confirm the target of a retrying "+
"delivery still exists; attempting anyway",
"delivery_id", task.DeliveryID,
"target_id", task.TargetID,
"error", err,
)
return false
}
targetType, reason := e.missingTargetReason(task.TargetID)
e.log.Warn(
"abandoning scheduled retry: target is gone",
"webhook_id", task.WebhookID,
"delivery_id", task.DeliveryID,
"target_id", task.TargetID,
"target_type", targetType,
)
e.failDelivery(webhookDB, d, targetType, reason)
return true
}
func (e *Engine) recoverInFlight(ctx context.Context) { func (e *Engine) recoverInFlight(ctx context.Context) {
var webhookIDs []string var webhookIDs []string
@@ -633,6 +696,20 @@ func (e *Engine) recoverSingleRetry(
) { ) {
target, err := e.loadTarget(d.TargetID) target, err := e.loadTarget(d.TargetID)
if err != nil { if err != nil {
// A target that is merely gone is an operator action with a
// terminal answer. Any other failure is the main database
// refusing to read, which is transient and must leave the
// delivery alone: failing every retrying delivery of every
// webhook on one bad read would be a far larger fault than
// the strand it is meant to clear.
if errors.Is(err, gorm.ErrRecordNotFound) {
e.failMissingTargetRetry(
webhookDB, webhookID, d,
)
return
}
e.log.Error( e.log.Error(
"failed to load target for retrying "+ "failed to load target for retrying "+
"delivery recovery", "delivery recovery",
@@ -1028,6 +1105,16 @@ func (e *Engine) sweepSingleRetry(
) { ) {
target, err := e.loadTarget(d.TargetID) target, err := e.loadTarget(d.TargetID)
if err != nil { if err != nil {
// Deleted is terminal, unreadable is not; see
// recoverSingleRetry.
if errors.Is(err, gorm.ErrRecordNotFound) {
e.failMissingTargetRetry(
webhookDB, webhookID, d,
)
return
}
e.log.Error( e.log.Error(
"retry sweep: failed to load target", "retry sweep: failed to load target",
"delivery_id", d.ID, "delivery_id", d.ID,
@@ -1134,6 +1221,113 @@ func (e *Engine) failUnretryableRetry(
target.Type, target.Type,
) )
e.failDelivery(webhookDB, d, target.Type, reason)
}
// failMissingTargetRetry terminally fails an orphaned retrying
// delivery whose target row is gone. Both restart recovery and the
// periodic sweep call it, so the transition exists once.
//
// Until it existed both paths logged the failed lookup and returned,
// which left the delivery retrying for the life of the database and
// the sweep repeating the same error every minute forever. Failing it
// with a recorded reason is the treatment the other orphaned-retry
// cases already get, so all of them read alike in the event log.
//
// Logged at warn rather than error: a deleted target is an operator
// action, not a system fault.
func (e *Engine) failMissingTargetRetry(
webhookDB *gorm.DB,
webhookID string,
d *database.Delivery,
) {
// Terminal, and reached from the recovery paths, so it takes
// ownership like every other write they make.
if !e.inflight.retainIdle(d.ID) {
return
}
defer e.inflight.release(d.ID)
targetType, reason := e.missingTargetReason(d.TargetID)
e.log.Warn(
"failing orphaned retrying delivery: "+
"its target no longer exists",
"webhook_id", webhookID,
"delivery_id", d.ID,
"target_id", d.TargetID,
"target_type", targetType,
)
e.failDelivery(webhookDB, d, targetType, reason)
}
// missingTargetReason describes a target id that no longer resolves,
// and returns the type of the deleted row where there still is one.
//
// The lookup is Unscoped because deletes are soft: the row survives
// with deleted_at set, invisible to loadTarget's default scope.
// Reading it is what separates "you deleted this target" from "this id
// never named a row" — different things to whoever reads the event
// log, and only the first is something an operator did. The widened
// scope is deliberately confined to this terminal path: the engine's
// normal target loading must go on refusing a deleted target, or
// deleting one would stop nothing.
//
// The type comes back so the caller can label the delivery's status
// transition with it. Where the row is gone entirely there is no type
// to give, and updateDeliveryStatus leaves the counter alone rather
// than opening a series named by the empty string.
func (e *Engine) missingTargetReason(
targetID string,
) (database.TargetType, string) {
var target database.Target
err := e.database.DB().Unscoped().
First(&target, "id = ?", targetID).Error
if err != nil {
return "", fmt.Sprintf(
"target %s no longer exists; the delivery "+
"cannot be retried and has been failed "+
"terminally",
targetID,
)
}
return target.Type, fmt.Sprintf(
"target %q (type %s) was deleted; the delivery "+
"cannot be retried and has been failed terminally",
target.Name, target.Type,
)
}
// failDelivery records why a delivery is over and then marks it
// failed. The caller must already own the delivery: every call site is
// either a worker holding the reference runTask took, or a recovery
// path that took one through retainIdle.
//
// The result row is written first and a failure to write it stops the
// transition, which is what keeps a delivery from ending failed with
// an empty event log — the state that leaves an operator with nothing
// but a server log line to work out what happened. A delivery whose
// reason could not be recorded stays in the non-terminal state it
// already holds, where the sweep will find it again; see
// bookkeepingFailed.
//
// The target type is a parameter rather than read off d because the
// orphaned-retry callers deliberately hold a delivery loaded without
// its Target relation: populating d.Target would make GORM's
// SaveBeforeAssociations upsert the whole target row — plaintext
// config, which for a slack target is the credential — into the
// per-webhook event database. See
// https://git.eeqj.de/sneak/webhooker/issues/206.
func (e *Engine) failDelivery(
webhookDB *gorm.DB,
d *database.Delivery,
targetType database.TargetType,
reason string,
) {
err := e.recordResult( err := e.recordResult(
webhookDB, webhookDB,
d, d,
@@ -1150,14 +1344,8 @@ func (e *Engine) failUnretryableRetry(
return return
} }
// The type is passed rather than assigned onto d: the delivery
// is loaded here without its target relation, and populating
// d.Target would make GORM's SaveBeforeAssociations upsert the
// whole target row — plaintext config, which for a slack target
// is the credential — into the per-webhook event database. See
// https://git.eeqj.de/sneak/webhooker/issues/206.
e.settleStatus( e.settleStatus(
webhookDB, d, target.Type, webhookDB, d, targetType,
database.DeliveryStatusFailed, database.DeliveryStatusFailed,
) )
} }
@@ -1178,9 +1366,19 @@ func (e *Engine) processDelivery(
"type", d.Target.Type, "type", d.Target.Type,
) )
e.settleStatus( // The reason is recorded, not just logged. This branch used
// to fail the delivery with no DeliveryResult at all, which
// showed in the event log as "failed, no attempts recorded
// yet" and left one server log line as the only account of
// why anywhere.
e.failDelivery(
webhookDB, d, d.Target.Type, webhookDB, d, d.Target.Type,
database.DeliveryStatusFailed, fmt.Sprintf(
"unknown target type %q: this build has no "+
"delivery implementation for it, so no "+
"attempt was made",
d.Target.Type,
),
) )
return return
@@ -1349,6 +1547,11 @@ func truncate(s string, maxLen int) string {
// --- Helper functions --- // --- Helper functions ---
// buildEventFromTask reconstructs the event a Task describes, as far
// as the Task itself goes. The fields it cannot fill — the body when
// it was too large to inline, and the receipt time, which no Task
// carries — come from the stored row in hydrateEvent, which every
// caller of this function runs next.
func buildEventFromTask(task *Task) database.Event { func buildEventFromTask(task *Task) database.Event {
event := database.Event{ event := database.Event{
EntrypointID: task.EntrypointID, EntrypointID: task.EntrypointID,
@@ -1376,28 +1579,66 @@ func buildTargetFromTask(task *Task) database.Target {
return target return target
} }
func (e *Engine) resolveEventBody( // hydrateEvent fills in the event fields a Task does not carry, by
// reading the stored event row.
//
// CreatedAt is the event's receipt time and lives only in that row.
// The Slack target renders it into every message it sends, so an
// unhydrated event puts the zero time in front of a human on every
// notification the product delivers. See
// https://git.eeqj.de/sneak/webhooker/issues/257.
//
// The body comes from the same row when the Task did not inline it,
// which is the case for a body at or above MaxInlineBodySize.
//
// A read failure is fatal to the delivery only when the body depended
// on it. When the Task inlined the body, the delivery has everything
// it needs to be sent and goes ahead with the timestamp unset: the row
// can be gone under a retention reap while a queued delivery still
// holds its body, and dropping a deliverable event to protect one
// metadata field would be a worse failure than the one it prevents.
func (e *Engine) hydrateEvent(
webhookDB *gorm.DB, webhookDB *gorm.DB,
event database.Event, event database.Event,
task *Task, task *Task,
) (database.Event, error) { ) (database.Event, error) {
if task.Body != nil { columns := []string{"created_at"}
event.Body = *task.Body
return event, nil if task.Body == nil {
columns = append(columns, "body")
} }
var dbEvent database.Event var dbEvent database.Event
err := webhookDB.Select("body"). err := webhookDB.Select(columns).
First(&dbEvent, "id = ?", task.EventID).Error First(&dbEvent, "id = ?", task.EventID).Error
if err != nil { if err != nil {
if task.Body == nil {
return event, fmt.Errorf( return event, fmt.Errorf(
"fetching event body: %w", err, "fetching event body: %w", err,
) )
} }
e.log.Warn(
"could not read the stored event; delivering "+
"the inlined body without its receipt time",
"event_id", task.EventID,
"delivery_id", task.DeliveryID,
"error", err,
)
event.Body = *task.Body
return event, nil
}
event.CreatedAt = dbEvent.CreatedAt
if task.Body != nil {
event.Body = *task.Body
} else {
event.Body = dbEvent.Body event.Body = dbEvent.Body
}
return event, nil return event, nil
} }

View File

@@ -377,6 +377,17 @@ func TestProcessRetryTask_SuccessfulRetry(t *testing.T) {
bodyStr := event.Body bodyStr := event.Body
cfg := iHTTPConfig(ts.URL) cfg := iHTTPConfig(ts.URL)
// The target row exists because the engine confirms a scheduled
// retry's target has not been deleted before it runs it. A retry
// task whose target id names no row at all is a state the service
// does not produce: the handler read that target to build the
// task. See https://git.eeqj.de/sneak/webhooker/issues/107.
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "retry-target",
database.TargetTypeHTTP, cfg, 5,
)
task := iTask( task := iTask(
d, event, s.WebhookID, targetID, d, event, s.WebhookID, targetID,
"retry-target", cfg, 5, 2, &bodyStr, "retry-target", cfg, 5, 2, &bodyStr,
@@ -456,6 +467,12 @@ func TestProcessRetryTask_LargeBody_FetchFromDB(
) )
cfg := iHTTPConfig(ts.URL) cfg := iHTTPConfig(ts.URL)
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "retry-large",
database.TargetTypeHTTP, cfg, 5,
)
task := iTask( task := iTask(
d, event, s.WebhookID, targetID, d, event, s.WebhookID, targetID,
"retry-large", cfg, 5, 2, nil, "retry-large", cfg, 5, 2, nil,
@@ -558,6 +575,12 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
bodyStr := event.Body bodyStr := event.Body
cfg := iHTTPConfig(ts.URL) cfg := iHTTPConfig(ts.URL)
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "retry-chan-test",
database.TargetTypeHTTP, cfg, 5,
)
task := iTask( task := iTask(
d, event, s.WebhookID, targetID, d, event, s.WebhookID, targetID,
"retry-chan-test", cfg, 5, 2, &bodyStr, "retry-chan-test", cfg, 5, 2, &bodyStr,

View File

@@ -96,7 +96,15 @@ func TestEventDBHoldsNoTargetRows(t *testing.T) {
) )
assertNoTargetRows(t, dbPath) assertNoTargetRows(t, dbPath)
// A retry. // A retry. Its target exists in the main database, because the
// engine confirms a scheduled retry's target has not been
// deleted before running it; see
// https://git.eeqj.de/sneak/webhooker/issues/107.
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "leaky-target",
database.TargetTypeHTTP, cfg, 5,
)
rd := iSeedDelivery( rd := iSeedDelivery(
t, s.WebhookDB, event.ID, targetID, t, s.WebhookDB, event.ID, targetID,
database.DeliveryStatusRetrying, database.DeliveryStatusRetrying,

View File

@@ -0,0 +1,442 @@
package delivery_test
import (
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/delivery"
)
// tsEventCreatedAt is the receipt time seeded on the events these
// tests deliver. It is far enough from both the zero time and from
// now that neither can be mistaken for it.
func tsEventCreatedAt() time.Time {
return time.Date(
2026, time.March, 4, 5, 6, 7, 0, time.UTC,
)
}
// tsZeroStamp is what a Slack message renders when the event handed
// to FormatSlackMessage carries no CreatedAt.
const tsZeroStamp = "*Timestamp:* `0001-01-01T00:00:00Z`"
// tsEventBody is the body seeded on every event in this file. It is
// small enough that a Task can inline it.
const tsEventBody = `{"hello":"world"}`
// tsUndeliverableHook stands in for a Slack incoming webhook on the
// tests that never send: the config parser requires a URL, but no
// request is made.
const tsUndeliverableHook = "https://hooks.slack.com/services/T/B/x"
// tsSink is a stand-in Slack incoming webhook that records the raw
// body posted to it.
type tsSink struct {
*httptest.Server
bodies chan []byte
}
func newTSSink(t *testing.T) *tsSink {
t.Helper()
s := &tsSink{bodies: make(chan []byte, 8)}
s.Server = httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
select {
case s.bodies <- body:
default:
}
w.WriteHeader(http.StatusOK)
},
))
t.Cleanup(s.Close)
return s
}
// text returns the Slack message text from the single payload the
// sink received.
func (s *tsSink) text(t *testing.T) string {
t.Helper()
select {
case raw := <-s.bodies:
t.Logf("raw slack payload: %s", raw)
var payload struct {
Text string `json:"text"`
}
require.NoError(t, json.Unmarshal(raw, &payload))
return payload.Text
case <-time.After(5 * time.Second):
t.Fatal("slack sink received no payload")
return ""
}
}
func tsSlackConfig(t *testing.T, url string) string {
t.Helper()
data, err := json.Marshal(
delivery.SlackTargetConfig{WebhookURL: url},
)
require.NoError(t, err)
return string(data)
}
// tsSeedEvent writes an event whose CreatedAt is tsEventCreatedAt
// rather than the write time, so an assertion on the rendered
// timestamp cannot pass by accident against "roughly now".
func tsSeedEvent(
t *testing.T, db *gorm.DB, webhookID string,
) database.Event {
t.Helper()
event := database.Event{
WebhookID: webhookID,
EntrypointID: uuid.New().String(),
Method: http.MethodPost,
Headers: `{}`,
Body: tsEventBody,
ContentType: "application/json",
}
event.ID = uuid.New().String()
event.CreatedAt = tsEventCreatedAt()
event.UpdatedAt = tsEventCreatedAt()
require.NoError(t, db.Create(&event).Error)
var stored database.Event
require.NoError(t,
db.First(&stored, "id = ?", event.ID).Error,
)
require.Equal(t,
tsEventCreatedAt().UTC(), stored.CreatedAt.UTC(),
"seeded created_at did not round-trip",
)
return event
}
// tsSeedTarget writes the slack target row into the main database.
// The retry path confirms the target still exists before sending.
func tsSeedTarget(
t *testing.T, mainDB *gorm.DB, webhookID, config string,
) database.Target {
t.Helper()
target := database.Target{
WebhookID: webhookID,
Name: "slack-sink",
Type: database.TargetTypeSlack,
Config: config,
Active: true,
}
require.NoError(t, mainDB.Create(&target).Error)
return target
}
func tsTask(
d database.Delivery,
event database.Event,
webhookID string,
target database.Target,
attemptNum int,
body *string,
) delivery.Task {
return delivery.Task{
DeliveryID: d.ID,
EventID: event.ID,
WebhookID: webhookID,
EntrypointID: event.EntrypointID,
TargetID: target.ID,
TargetName: target.Name,
TargetType: database.TargetTypeSlack,
TargetConfig: target.Config,
MaxRetries: 0,
Method: event.Method,
Headers: event.Headers,
ContentType: event.ContentType,
Body: body,
AttemptNum: attemptNum,
}
}
func tsAssertRealTimestamp(t *testing.T, text string) {
t.Helper()
assert.NotContains(t, text, tsZeroStamp,
"slack message carries the zero timestamp",
)
assert.Contains(t, text,
"*Timestamp:* `"+
tsEventCreatedAt().UTC().Format(time.RFC3339)+"`",
"slack message does not carry the event's receipt time",
)
}
// tsCase is one end-to-end delivery of a seeded event to a slack
// sink, over whichever engine path `process` names.
type tsCase struct {
// status is the delivery row's status before the engine runs.
// The retry path refuses a delivery that is not retrying.
status database.DeliveryStatus
// inlineBody mirrors a Task built for a body under
// MaxInlineBodySize. When false the engine reads the body back
// from the stored row.
inlineBody bool
attemptNum int
process func(
ctx context.Context, e *delivery.Engine, task *delivery.Task,
)
}
// run delivers one event through the named path and returns the
// Slack message text the sink received.
func (c tsCase) run(t *testing.T) (iSetup, database.Delivery, string) {
t.Helper()
s := newISetup(t)
sink := newTSSink(t)
cfg := tsSlackConfig(t, sink.URL)
target := tsSeedTarget(t, s.MainDB, s.WebhookID, cfg)
event := tsSeedEvent(t, s.WebhookDB, s.WebhookID)
d := iSeedDelivery(
t, s.WebhookDB, event.ID, target.ID, c.status,
)
var body *string
if c.inlineBody {
bodyStr := event.Body
body = &bodyStr
}
task := tsTask(
d, event, s.WebhookID, target, c.attemptNum, body,
)
c.process(context.TODO(), s.Engine, &task)
return s, d, sink.text(t)
}
// TestSlackFirstAttemptCarriesEventTimestamp covers the path an
// event takes on its first delivery: the task comes from the
// receiver and the engine reconstructs the event from it.
func TestSlackFirstAttemptCarriesEventTimestamp(t *testing.T) {
t.Parallel()
s, d, text := tsCase{
status: database.DeliveryStatusPending,
inlineBody: true,
attemptNum: 1,
process: func(
ctx context.Context,
e *delivery.Engine,
task *delivery.Task,
) {
e.ExportProcessNewTask(ctx, task)
},
}.run(t)
tsAssertRealTimestamp(t, text)
iAssertStatus(t, s.WebhookDB, d.ID,
database.DeliveryStatusDelivered,
)
}
// TestSlackFirstAttemptLargeBodyCarriesEventTimestamp covers the
// first-attempt path for an event whose body exceeded
// MaxInlineBodySize, so the task carries no body and the engine
// reads it back from the stored row.
func TestSlackFirstAttemptLargeBodyCarriesEventTimestamp(
t *testing.T,
) {
t.Parallel()
_, _, text := tsCase{
status: database.DeliveryStatusPending,
inlineBody: false,
attemptNum: 1,
process: func(
ctx context.Context,
e *delivery.Engine,
task *delivery.Task,
) {
e.ExportProcessNewTask(ctx, task)
},
}.run(t)
tsAssertRealTimestamp(t, text)
}
// TestSlackRetryCarriesEventTimestamp covers the retry path, which
// reconstructs the event from the same task the first attempt used.
func TestSlackRetryCarriesEventTimestamp(t *testing.T) {
t.Parallel()
s, d, text := tsCase{
status: database.DeliveryStatusRetrying,
inlineBody: true,
attemptNum: 2,
process: func(
ctx context.Context,
e *delivery.Engine,
task *delivery.Task,
) {
e.ExportProcessRetryTask(ctx, task)
},
}.run(t)
tsAssertRealTimestamp(t, text)
iAssertStatus(t, s.WebhookDB, d.ID,
database.DeliveryStatusDelivered,
)
}
// TestFormatSlackMessageOverTaskReconstructedEvent asserts on the
// formatted message directly, over the event the delivery paths
// reconstruct from a Task. It is the unit-level guard under the
// end-to-end tests: revert the CreatedAt population in hydrateEvent
// and this fails on the zero timestamp.
func TestFormatSlackMessageOverTaskReconstructedEvent(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
cfg := tsSlackConfig(t, tsUndeliverableHook)
target := tsSeedTarget(t, s.MainDB, s.WebhookID, cfg)
event := tsSeedEvent(t, s.WebhookDB, s.WebhookID)
d := iSeedDelivery(
t, s.WebhookDB, event.ID, target.ID,
database.DeliveryStatusPending,
)
bodyStr := event.Body
task := tsTask(d, event, s.WebhookID, target, 1, &bodyStr)
rebuilt, err := s.Engine.ExportEventForTask(
s.WebhookDB, &task,
)
require.NoError(t, err)
assert.False(t, rebuilt.CreatedAt.IsZero(),
"reconstructed event carries the zero time",
)
assert.Equal(t,
tsEventCreatedAt().UTC(), rebuilt.CreatedAt.UTC(),
)
tsAssertRealTimestamp(
t, delivery.FormatSlackMessage(&rebuilt),
)
}
// TestFormatSlackMessageZeroTimestamp asserts the rendering choice
// directly, without going through the engine: a zero CreatedAt (the
// shape a reaped-row fallback produces) renders as "unknown" rather
// than the year-1 zero time, while a real CreatedAt still renders as
// RFC3339.
func TestFormatSlackMessageZeroTimestamp(t *testing.T) {
t.Parallel()
zeroEvent := database.Event{
Method: http.MethodPost,
ContentType: testContentType,
Body: tsEventBody,
}
zeroText := delivery.FormatSlackMessage(&zeroEvent)
assert.NotContains(t, zeroText, "0001-01-01",
"slack message carries the zero-time year",
)
assert.Contains(t, zeroText, "*Timestamp:* `unknown`",
"slack message does not mark an unset receipt time as unknown",
)
nonZeroEvent := zeroEvent
nonZeroEvent.CreatedAt = tsEventCreatedAt()
nonZeroText := delivery.FormatSlackMessage(&nonZeroEvent)
assert.Contains(t, nonZeroText,
"*Timestamp:* `"+
tsEventCreatedAt().UTC().Format(time.RFC3339)+"`",
"slack message does not render a real receipt time as RFC3339",
)
}
// TestEventReconstructionSurvivesAReapedRow pins the fallback: an
// event row reaped by retention while its delivery still holds the
// body inline is still delivered, with the receipt time unset,
// rather than dropped.
func TestEventReconstructionSurvivesAReapedRow(t *testing.T) {
t.Parallel()
s := newISetup(t)
cfg := tsSlackConfig(t, tsUndeliverableHook)
target := tsSeedTarget(t, s.MainDB, s.WebhookID, cfg)
event := tsSeedEvent(t, s.WebhookDB, s.WebhookID)
d := iSeedDelivery(
t, s.WebhookDB, event.ID, target.ID,
database.DeliveryStatusPending,
)
bodyStr := event.Body
task := tsTask(d, event, s.WebhookID, target, 1, &bodyStr)
require.NoError(t, s.WebhookDB.Unscoped().Delete(
&database.Event{}, "id = ?", event.ID,
).Error)
rebuilt, err := s.Engine.ExportEventForTask(
s.WebhookDB, &task,
)
require.NoError(t, err)
assert.Equal(t, bodyStr, rebuilt.Body)
assert.True(t, rebuilt.CreatedAt.IsZero())
// A task with no inlined body has nothing left to deliver, so
// the same reaped row is an error there.
noBody := task
noBody.Body = nil
_, err = s.Engine.ExportEventForTask(s.WebhookDB, &noBody)
require.Error(t, err)
}

View File

@@ -151,6 +151,16 @@ func (e *Engine) ExportProcessRetryTask(
e.processRetryTask(ctx, task) e.processRetryTask(ctx, task)
} }
// ExportEventForTask exposes the event reconstruction the delivery
// paths run: buildEventFromTask followed by hydrateEvent.
func (e *Engine) ExportEventForTask(
webhookDB *gorm.DB, task *Task,
) (database.Event, error) {
return e.hydrateEvent(
webhookDB, buildEventFromTask(task), task,
)
}
// ExportProcessDelivery exposes processDelivery. // ExportProcessDelivery exposes processDelivery.
func (e *Engine) ExportProcessDelivery( func (e *Engine) ExportProcessDelivery(
ctx context.Context, ctx context.Context,

View File

@@ -229,6 +229,13 @@ func mExhaustRetries(t *testing.T, s iSetup) {
body := event.Body body := event.Body
cfg := iHTTPConfig(ts.URL) cfg := iHTTPConfig(ts.URL)
// The retry below is only run if its target still exists; see
// https://git.eeqj.de/sneak/webhooker/issues/107.
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "metrics-fail",
database.TargetTypeHTTP, cfg, 2,
)
first := iTask( first := iTask(
d, event, s.WebhookID, targetID, d, event, s.WebhookID, targetID,
"metrics-fail", cfg, 2, 1, &body, "metrics-fail", cfg, 2, 1, &body,
@@ -289,6 +296,13 @@ func TestDeliveryMetrics_CircuitBreakerGauge(t *testing.T) {
// rather than the budget is what stops the delivery. // rather than the budget is what stops the delivery.
maxRetries := delivery.ExportDefaultFailureThreshold + 5 maxRetries := delivery.ExportDefaultFailureThreshold + 5
// The retries below are only run if their target still exists;
// see https://git.eeqj.de/sneak/webhooker/issues/107.
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "metrics-trip",
database.TargetTypeHTTP, cfg, maxRetries,
)
first := iTask( first := iTask(
d, event, s.WebhookID, targetID, d, event, s.WebhookID, targetID,
"metrics-trip", cfg, maxRetries, 1, &body, "metrics-trip", cfg, maxRetries, 1, &body,
@@ -353,6 +367,11 @@ func TestDeliveryMetrics_BreakerBlockedIsNotAnAttempt(
cfg := iHTTPConfig(ts.URL) cfg := iHTTPConfig(ts.URL)
maxRetries := delivery.ExportDefaultFailureThreshold + 5 maxRetries := delivery.ExportDefaultFailureThreshold + 5
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "metrics-blocked",
database.TargetTypeHTTP, cfg, maxRetries,
)
first := iTask( first := iTask(
d, event, s.WebhookID, targetID, d, event, s.WebhookID, targetID,
"metrics-blocked", cfg, maxRetries, 1, &body, "metrics-blocked", cfg, maxRetries, 1, &body,

View File

@@ -231,10 +231,15 @@ func FormatSlackMessage(
event.ContentType, event.ContentType,
) )
timestamp := "unknown"
if !event.CreatedAt.IsZero() {
timestamp = event.CreatedAt.UTC().Format(time.RFC3339)
}
fmt.Fprintf( fmt.Fprintf(
&b, &b,
"*Timestamp:* `%s`\n", "*Timestamp:* `%s`\n",
event.CreatedAt.UTC().Format(time.RFC3339), timestamp,
) )
fmt.Fprintf( fmt.Fprintf(

View File

@@ -0,0 +1,531 @@
package delivery_test
import (
"context"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/delivery"
)
// The two terminal-state gaps of
// https://git.eeqj.de/sneak/webhooker/issues/107: a delivery failed
// with nothing in its event log to say why, and a retrying delivery
// whose target was deleted, which used to keep sending and then never
// terminalise.
// tUnknownType is a target type no build implements. It stands in for
// a target whose type was written by a build that knew a type this one
// does not.
const tUnknownType = database.TargetType("pubsub")
// tSeedDeletedTarget creates a target, a retrying delivery against it
// with one recorded failed attempt, and then deletes the target the
// way the source page does.
//
// It asserts the delete is soft, because that is the whole reason the
// engine could not tell a deleted target from a target id that never
// named a row: the surviving row is invisible to a scoped read.
func tSeedDeletedTarget(
t *testing.T,
s iSetup,
name, url string,
) string {
t.Helper()
targetID := uuid.New().String()
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, name,
database.TargetTypeHTTP, iHTTPConfig(url), 5,
)
event := iSeedEvent(
t, s.WebhookDB, s.WebhookID, `{"target":"deleted"}`,
)
d := iSeedDelivery(
t, s.WebhookDB, event.ID, targetID,
database.DeliveryStatusRetrying,
)
iSeedFailedResult(t, s.WebhookDB, d.ID)
require.NoError(t, s.MainDB.Delete(
&database.Target{}, "id = ?", targetID,
).Error)
var scoped, unscoped int64
require.NoError(t, s.MainDB.
Model(&database.Target{}).
Where("id = ?", targetID).
Count(&scoped).Error)
require.NoError(t, s.MainDB.Unscoped().
Model(&database.Target{}).
Where("id = ?", targetID).
Count(&unscoped).Error)
require.Zero(t, scoped,
"the deleted target is still visible to a scoped read",
)
require.Equal(t, int64(1), unscoped,
"the delete was hard, so this test proves nothing about "+
"the soft-delete case it exists for",
)
return d.ID
}
// tLastResult returns a delivery's final recorded attempt, asserting
// the expected number of them.
func tLastResult(
t *testing.T,
s iSetup,
deliveryID string,
want int,
) database.DeliveryResult {
t.Helper()
results := iResults(t, s.WebhookDB, deliveryID)
require.Len(t, results, want)
return results[want-1]
}
// --- 1. A failure with nothing recorded ---
func TestProcessDelivery_UnknownTargetType_RecordsWhy(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
targetID := uuid.New().String()
event := iSeedEvent(
t, s.WebhookDB, s.WebhookID, `{"unknown":"type"}`,
)
seeded := iSeedDelivery(
t, s.WebhookDB, event.ID, targetID,
database.DeliveryStatusPending,
)
target := database.Target{
Name: "mystery",
Type: tUnknownType,
Config: iHTTPConfig("http://example.com/hook"),
}
target.ID = targetID
d := database.Delivery{
EventID: event.ID,
TargetID: targetID,
Status: database.DeliveryStatusPending,
Event: event,
Target: target,
}
d.ID = seeded.ID
body := event.Body
task := iTask(
seeded, event, s.WebhookID, targetID, "mystery",
target.Config, 0, 1, &body,
)
task.TargetType = tUnknownType
s.Engine.ExportProcessDelivery(
context.Background(), s.WebhookDB, &d, &task,
)
iAssertStatus(
t, s.WebhookDB, d.ID, database.DeliveryStatusFailed,
)
last := tLastResult(t, s, d.ID, 1)
assert.False(t, last.Success)
assert.Equal(t, 1, last.AttemptNum)
assert.Contains(t, last.Error, string(tUnknownType),
"the recorded reason does not name the offending type",
)
}
// --- 2. A retrying delivery whose target is gone ---
func TestRecoverSingleRetry_TargetDeleted(t *testing.T) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "deleted-target-recovery",
)
deliveryID := tSeedDeletedTarget(
t, s, "gone-on-recovery", "http://example.com/hook",
)
s.Engine.ExportRecoverWebhookDeliveries(
context.Background(), s.WebhookID,
)
iAssertStatus(
t, s.WebhookDB, deliveryID,
database.DeliveryStatusFailed,
)
last := tLastResult(t, s, deliveryID, 2)
assert.False(t, last.Success)
assert.Equal(t, 2, last.AttemptNum)
assert.Contains(t, last.Error, "gone-on-recovery")
assert.Contains(t, last.Error, "was deleted")
assert.Empty(t, s.Engine.ExportRetryCh(),
"a delivery whose target is gone was rescheduled",
)
assert.Zero(t, s.Engine.ExportInflightHeld(),
"the terminal path leaked its ownership reference",
)
}
func TestSweepSingleRetry_TargetDeleted(t *testing.T) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "deleted-target-sweep",
)
deliveryID := tSeedDeletedTarget(
t, s, "gone-on-sweep", "http://example.com/hook",
)
// Twice, because the bug was an error the sweep repeated every
// minute for the life of the database: the second sweep must
// find nothing left to do.
s.Engine.ExportSweepWebhookRetries(
context.Background(), s.WebhookID,
)
s.Engine.ExportSweepWebhookRetries(
context.Background(), s.WebhookID,
)
iAssertStatus(
t, s.WebhookDB, deliveryID,
database.DeliveryStatusFailed,
)
last := tLastResult(t, s, deliveryID, 2)
assert.Contains(t, last.Error, "gone-on-sweep")
assert.Contains(t, last.Error, "was deleted")
assert.Empty(t, s.Engine.ExportRetryCh())
assert.Zero(t, s.Engine.ExportInflightHeld())
}
// TestSweepSingleRetry_TargetNeverExisted covers the other half of the
// soft-delete distinction: an id with no row at all, deleted or
// otherwise, must not be reported as something the operator deleted.
func TestSweepSingleRetry_TargetNeverExisted(t *testing.T) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "target-never-existed",
)
targetID := uuid.New().String()
event := iSeedEvent(
t, s.WebhookDB, s.WebhookID, `{"target":"absent"}`,
)
d := iSeedDelivery(
t, s.WebhookDB, event.ID, targetID,
database.DeliveryStatusRetrying,
)
iSeedFailedResult(t, s.WebhookDB, d.ID)
s.Engine.ExportSweepWebhookRetries(
context.Background(), s.WebhookID,
)
iAssertStatus(
t, s.WebhookDB, d.ID, database.DeliveryStatusFailed,
)
last := tLastResult(t, s, d.ID, 2)
assert.Contains(t, last.Error, targetID)
assert.Contains(t, last.Error, "no longer exists")
assert.NotContains(t, last.Error, "was deleted",
"an id that never named a row was reported as a deletion",
)
}
// TestFailMissingTargetRetry_WritesNoTargetRow holds the new terminal
// path to the same rule as the existing one: no target row, and so no
// plaintext target config, may be written into the per-webhook event
// database. See https://git.eeqj.de/sneak/webhooker/issues/206.
func TestFailMissingTargetRetry_WritesNoTargetRow(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "no-target-row-deleted",
)
hookURL := "https://hooks.slack.com/services/T00/B00/x"
deliveryID := tSeedDeletedTarget(
t, s, "credential-bearing", hookURL,
)
s.Engine.ExportSweepWebhookRetries(
context.Background(), s.WebhookID,
)
iAssertStatus(
t, s.WebhookDB, deliveryID,
database.DeliveryStatusFailed,
)
var configs []string
require.NoError(t, s.WebhookDB.
Table("targets").
Pluck("config", &configs).Error)
assert.Empty(t, configs,
"the deleted-target terminal path wrote a target row "+
"into the per-webhook event database",
)
}
// --- 3. The scheduled retry chain ---
// tRetryChainSetup wires a counting sink and a retrying delivery
// against a live target pointing at it, and returns the task a
// scheduled retry would carry — config and all, snapshotted as
// ScheduleRetry snapshots it.
func tRetryChainSetup(
t *testing.T,
s iSetup,
name string,
hits *atomic.Int64,
) (delivery.Task, string) {
t.Helper()
ts := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
hits.Add(1)
w.WriteHeader(http.StatusOK)
},
))
t.Cleanup(ts.Close)
iCreateWebhook(t, s.MainDB, s.WebhookID, name)
targetID := uuid.New().String()
cfg := iHTTPConfig(ts.URL)
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, name,
database.TargetTypeHTTP, cfg, 5,
)
event := iSeedEvent(
t, s.WebhookDB, s.WebhookID, `{"chain":"retry"}`,
)
d := iSeedDelivery(
t, s.WebhookDB, event.ID, targetID,
database.DeliveryStatusRetrying,
)
iSeedFailedResult(t, s.WebhookDB, d.ID)
body := event.Body
return iTask(
d, event, s.WebhookID, targetID, name, cfg, 5, 2, &body,
), targetID
}
// TestProcessRetryTask_TargetDeleted_MakesNoAttempt is the half the
// deployability audit found worse than filed: terminalising on
// recovery and sweep alone leaves the already-scheduled timer chain
// running, and it holds the target's configuration from before the
// deletion, so it goes on sending to a destination that was removed.
func TestProcessRetryTask_TargetDeleted_MakesNoAttempt(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
var hits atomic.Int64
task, targetID := tRetryChainSetup(
t, s, "gone-mid-chain", &hits,
)
require.NoError(t, s.MainDB.Delete(
&database.Target{}, "id = ?", targetID,
).Error)
s.Engine.ExportProcessRetryTask(
context.Background(), &task,
)
assert.Zero(t, hits.Load(),
"a scheduled retry fired at a target the operator "+
"had already deleted",
)
iAssertStatus(
t, s.WebhookDB, task.DeliveryID,
database.DeliveryStatusFailed,
)
last := tLastResult(t, s, task.DeliveryID, 2)
assert.False(t, last.Success)
assert.Contains(t, last.Error, "was deleted")
assert.Zero(t, s.Engine.ExportInflightHeld())
}
// TestProcessRetryTask_TargetPresent_StillDelivers is the guard's
// mutation check: a liveness check that refused every retry would pass
// the test above and break every retry there is.
func TestProcessRetryTask_TargetPresent_StillDelivers(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
var hits atomic.Int64
task, _ := tRetryChainSetup(t, s, "still-there", &hits)
s.Engine.ExportProcessRetryTask(
context.Background(), &task,
)
assert.Equal(t, int64(1), hits.Load())
iAssertStatus(
t, s.WebhookDB, task.DeliveryID,
database.DeliveryStatusDelivered,
)
}
// TestProcessRetryTask_TargetUnreadable_StillDelivers pins the other
// half of the guard: only a target that is confirmed gone stops a
// retry. A main database that cannot be read is a transient fault, and
// a guard that abandoned deliveries on one would be a worse bug than
// the one it fixes.
func TestProcessRetryTask_TargetUnreadable_StillDelivers(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
var hits atomic.Int64
task, _ := tRetryChainSetup(t, s, "unreadable-main", &hits)
sqlDB, err := s.MainDB.DB()
require.NoError(t, err)
require.NoError(t, sqlDB.Close())
s.Engine.ExportProcessRetryTask(
context.Background(), &task,
)
assert.Equal(t, int64(1), hits.Load(),
"a retry was abandoned because the main database "+
"could not be read, not because its target was gone",
)
iAssertStatus(
t, s.WebhookDB, task.DeliveryID,
database.DeliveryStatusDelivered,
)
}
// TestRecoverSingleRetry_TargetUnreadable_LeavesDeliveryAlone is the
// same rule on the recovery path. A read failure that is not
// "record not found" must leave every retrying delivery of every
// webhook exactly as it was.
func TestRecoverSingleRetry_TargetUnreadable_LeavesDeliveryAlone(
t *testing.T,
) {
t.Parallel()
s := newISetup(t)
iCreateWebhook(
t, s.MainDB, s.WebhookID, "unreadable-on-recovery",
)
targetID := uuid.New().String()
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "healthy",
database.TargetTypeHTTP,
iHTTPConfig("http://example.com/hook"), 5,
)
event := iSeedEvent(
t, s.WebhookDB, s.WebhookID, `{"still":"retrying"}`,
)
d := iSeedDelivery(
t, s.WebhookDB, event.ID, targetID,
database.DeliveryStatusRetrying,
)
iSeedFailedResult(t, s.WebhookDB, d.ID)
sqlDB, err := s.MainDB.DB()
require.NoError(t, err)
require.NoError(t, sqlDB.Close())
s.Engine.ExportRecoverRetryingDeliveries(
s.WebhookDB, s.WebhookID,
)
iAssertStatus(
t, s.WebhookDB, d.ID,
database.DeliveryStatusRetrying,
)
assert.Len(t, iResults(t, s.WebhookDB, d.ID), 1,
"an unreadable main database produced a terminal "+
"failure row",
)
assert.Zero(t, s.Engine.ExportInflightHeld())
}