Compare commits
1 Commits
issue-97-l
...
issue-80-c
| Author | SHA1 | Date | |
|---|---|---|---|
| 985464dcf9 |
17
README.md
17
README.md
@@ -89,10 +89,27 @@ TTY detection, and security headers are always applied.
|
||||
| `PORT` | HTTP listen port | `8080` |
|
||||
| `DATA_DIR` | Directory for all SQLite databases | `/var/lib/webhooker` |
|
||||
| `DEBUG` | Enable debug logging | `false` |
|
||||
| `MAINTENANCE_MODE` | Serve the maintenance page | `false` |
|
||||
| `METRICS_USERNAME` | Basic auth username for `/metrics` | `""` |
|
||||
| `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` |
|
||||
| `SENTRY_DSN` | Sentry error reporting DSN | `""` |
|
||||
|
||||
#### Invalid values abort startup
|
||||
|
||||
The defaults above apply **only** to variables that are unset (or set
|
||||
to an empty string). A variable that is set but cannot be parsed is a
|
||||
fatal configuration error: webhooker logs the offending variable and
|
||||
its value and refuses to start, rather than silently running with a
|
||||
substituted default. `PORT=eighty`, `DEBUG=ture`, and
|
||||
`RETENTION_SWEEP_INTERVAL=1 hour` all abort startup. `PORT` must
|
||||
additionally be a number in the range 1–65535.
|
||||
|
||||
Boolean variables (`DEBUG`, `MAINTENANCE_MODE`) accept exactly the
|
||||
spellings Go's `strconv.ParseBool` accepts — `1`, `t`, `T`, `TRUE`,
|
||||
`true`, `True`, `0`, `f`, `F`, `FALSE`, `false`, `False` — and nothing
|
||||
else. `yes`, `on`, and `off` are rejected rather than quietly treated
|
||||
as false.
|
||||
|
||||
On first startup, webhooker automatically generates a cryptographically
|
||||
secure session encryption key and stores it in the database. This key
|
||||
persists across restarts — no manual key management is needed.
|
||||
|
||||
34
TODO.md
34
TODO.md
@@ -10,30 +10,28 @@
|
||||
|
||||
# Status
|
||||
|
||||
pre-1.0. No git tags exist. main (afe88c6) is a working webhook proxy
|
||||
pre-1.0. No git tags exist. main (4f5ecb1) is a working webhook proxy
|
||||
with auth, CSRF/SSRF protections, login rate limiting, Slack target,
|
||||
policy compliance (#6), and pinned lint tooling (#55). Note: TODO.md was
|
||||
deliberately deleted from this repo in f9a9569 (2026-03-01, #6); its
|
||||
content was folded into the README TODO section, which this draft
|
||||
reconstructs as of 2026-07-06.
|
||||
policy compliance (#6), pinned lint tooling (#55), a per-webhook event
|
||||
retention reaper (#63), and fail-loud configuration parsing (#80). Note:
|
||||
TODO.md was deliberately deleted from this repo in f9a9569 (2026-03-01,
|
||||
#6); its content was folded into the README TODO section, which this
|
||||
draft reconstructs as of 2026-07-06.
|
||||
|
||||
# Next Step
|
||||
|
||||
Implement automatic event retention cleanup based on retention_days: a
|
||||
periodic maintenance job that deletes Events, Deliveries, and
|
||||
DeliveryResults older than the parent webhook's retention_days from each
|
||||
per-webhook event database. The field exists on the Webhook model and
|
||||
the README promises the behavior, but nothing enforces it, so event
|
||||
databases currently grow without bound.
|
||||
Manual event redelivery from the web UI (replay is a core promised
|
||||
capability in the README rationale).
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-08-09 Root the delivery engine's worker pool and the retention
|
||||
reaper's sweep loop at `context.Background()` rather than the fx
|
||||
`OnStart` hook context (#97), which carries fx's 15s start timeout and
|
||||
killed both roughly fifteen seconds after boot: the proxy silently
|
||||
stopped delivering webhooks entirely, and the reaper never ran a
|
||||
single sweep under its default one-hour interval
|
||||
- 2026-08-09 Configuration parsing fails loudly on set-but-unparseable
|
||||
environment values: `envInt` removed in favour of `envPositiveInt`
|
||||
plus a `PORT` range check, `envBool` now parses with
|
||||
`strconv.ParseBool`, and defaults apply only to unset variables (#80)
|
||||
- 2026-08-07 Automatic event retention cleanup based on
|
||||
`retention_days`, deleting expired events, deliveries, and delivery
|
||||
results from each per-webhook event database (#63)
|
||||
- 2026-08-07 Update golangci-lint to v2.12.2 (Docker image digest in
|
||||
`Dockerfile`, release-archive sha256 pins in `script/bootstrap`),
|
||||
adopt the canonical `.golangci.yml` (v2 `linters.settings` layout so
|
||||
@@ -62,8 +60,6 @@ databases currently grow without bound.
|
||||
|
||||
# Future Steps
|
||||
|
||||
- Manual event redelivery from the web UI (replay is a core promised
|
||||
capability in the README rationale)
|
||||
- Delivery status and retry management UI
|
||||
- Per-webhook rate limiting in the receiver handler (per-webhook config
|
||||
plus handler enforcement; global limits must not apply to receiver
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"go.uber.org/fx"
|
||||
@@ -31,12 +30,24 @@ const (
|
||||
// defaultRetentionSweepInterval is how often the retention
|
||||
// reaper deletes events older than each webhook's RetentionDays.
|
||||
defaultRetentionSweepInterval = time.Hour
|
||||
|
||||
// maxPort is the highest valid TCP port number. The lower
|
||||
// bound (at least 1) is enforced by envPositiveInt.
|
||||
maxPort = 65535
|
||||
)
|
||||
|
||||
// ErrInvalidEnvironment is returned when WEBHOOKER_ENVIRONMENT
|
||||
// contains an unrecognised value.
|
||||
var ErrInvalidEnvironment = errors.New("invalid environment")
|
||||
|
||||
// ErrNonPositiveValue is returned when an environment variable that
|
||||
// requires a positive integer is set to zero or a negative number.
|
||||
var ErrNonPositiveValue = errors.New("value must be positive")
|
||||
|
||||
// ErrInvalidPort is returned when an environment variable holding a
|
||||
// TCP port number is set above the valid port range.
|
||||
var ErrInvalidPort = errors.New("invalid port")
|
||||
|
||||
//nolint:revive // ConfigParams is a standard fx naming convention.
|
||||
type ConfigParams struct {
|
||||
fx.In
|
||||
@@ -81,27 +92,81 @@ func envString(key string) string {
|
||||
}
|
||||
|
||||
// envBool returns the value of the named environment variable
|
||||
// parsed as a boolean. Returns defaultValue if not set.
|
||||
func envBool(key string, defaultValue bool) bool {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
return strings.EqualFold(v, "true") || v == "1"
|
||||
// parsed as a boolean. Returns defaultValue if not set. If the
|
||||
// variable is set but cannot be parsed, it returns a wrapped error
|
||||
// naming the key and the bad value, so startup fails loudly rather
|
||||
// than silently falling back to the default.
|
||||
//
|
||||
// Parsing is strconv.ParseBool, which accepts 1, t, T, TRUE, true,
|
||||
// True, 0, f, F, FALSE, false and False. Anything else — "yes",
|
||||
// "on", or a typo like "ture" — is an error rather than a silent
|
||||
// false.
|
||||
func envBool(key string, defaultValue bool) (bool, error) {
|
||||
v := os.Getenv(key)
|
||||
if v == "" {
|
||||
return defaultValue, nil
|
||||
}
|
||||
|
||||
return defaultValue
|
||||
b, err := strconv.ParseBool(v)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf(
|
||||
"invalid boolean for %s: %q: %w", key, v, err,
|
||||
)
|
||||
}
|
||||
|
||||
return b, nil
|
||||
}
|
||||
|
||||
// envPositiveInt returns the value of the named environment variable
|
||||
// parsed as a positive integer. Returns defaultValue if not set. If
|
||||
// the variable is set but cannot be parsed, or parses to less than
|
||||
// one, it returns a wrapped error naming the key and the bad value,
|
||||
// so startup fails loudly rather than silently falling back to the
|
||||
// default.
|
||||
func envPositiveInt(
|
||||
key string,
|
||||
defaultValue int,
|
||||
) (int, error) {
|
||||
v := os.Getenv(key)
|
||||
if v == "" {
|
||||
return defaultValue, nil
|
||||
}
|
||||
|
||||
// envInt returns the value of the named environment variable
|
||||
// parsed as an integer. Returns defaultValue if not set or
|
||||
// unparseable.
|
||||
func envInt(key string, defaultValue int) int {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
i, err := strconv.Atoi(v)
|
||||
if err == nil {
|
||||
return i
|
||||
}
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf(
|
||||
"invalid integer for %s: %q: %w", key, v, err,
|
||||
)
|
||||
}
|
||||
|
||||
return defaultValue
|
||||
if i < 1 {
|
||||
return 0, fmt.Errorf(
|
||||
"%w: %s must be at least 1, got %q",
|
||||
ErrNonPositiveValue, key, v,
|
||||
)
|
||||
}
|
||||
|
||||
return i, nil
|
||||
}
|
||||
|
||||
// envPort returns the value of the named environment variable parsed
|
||||
// as a TCP port number. Returns defaultValue if not set. A set value
|
||||
// that is unparseable, below 1, or above maxPort is a hard error
|
||||
// naming the key and the bad value.
|
||||
func envPort(key string, defaultValue int) (int, error) {
|
||||
port, err := envPositiveInt(key, defaultValue)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
if port > maxPort {
|
||||
return 0, fmt.Errorf(
|
||||
"%w: %s must be at most %d, got %d",
|
||||
ErrInvalidPort, key, maxPort, port,
|
||||
)
|
||||
}
|
||||
|
||||
return port, nil
|
||||
}
|
||||
|
||||
// envDuration returns the value of the named environment variable
|
||||
@@ -128,32 +193,52 @@ func envDuration(
|
||||
return d, nil
|
||||
}
|
||||
|
||||
// New creates a Config by reading environment variables.
|
||||
//
|
||||
//nolint:revive // lc parameter is required by fx even if unused.
|
||||
func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
|
||||
log := params.Logger.Get()
|
||||
|
||||
// Determine environment from WEBHOOKER_ENVIRONMENT env var,
|
||||
// default to dev
|
||||
// resolveEnvironment reads WEBHOOKER_ENVIRONMENT, defaulting to
|
||||
// dev, and rejects unrecognised values.
|
||||
func resolveEnvironment() (string, error) {
|
||||
environment := os.Getenv("WEBHOOKER_ENVIRONMENT")
|
||||
if environment == "" {
|
||||
environment = EnvironmentDev
|
||||
}
|
||||
|
||||
// Validate environment
|
||||
if environment != EnvironmentDev &&
|
||||
environment != EnvironmentProd {
|
||||
return nil, fmt.Errorf(
|
||||
return "", fmt.Errorf(
|
||||
"%w: WEBHOOKER_ENVIRONMENT must be '%s' or '%s', got '%s'",
|
||||
ErrInvalidEnvironment,
|
||||
EnvironmentDev, EnvironmentProd, environment,
|
||||
)
|
||||
}
|
||||
|
||||
// Parse the retention sweep interval; a set-but-unparseable value
|
||||
// is a hard error so fx aborts startup rather than silently using
|
||||
// the default.
|
||||
return environment, nil
|
||||
}
|
||||
|
||||
// loadFromEnv builds a Config from the environment. Every value that
|
||||
// needs parsing fails loudly when it is set but unparseable: the
|
||||
// documented defaults apply only to variables that are unset (or
|
||||
// empty), never as a substitute for a value the operator actually
|
||||
// provided.
|
||||
func loadFromEnv() (*Config, error) {
|
||||
environment, err := resolveEnvironment()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
port, err := envPort("PORT", defaultPort)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
debug, err := envBool("DEBUG", false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
maintenanceMode, err := envBool("MAINTENANCE_MODE", false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
retentionSweepInterval, err := envDuration(
|
||||
"RETENTION_SWEEP_INTERVAL",
|
||||
defaultRetentionSweepInterval,
|
||||
@@ -162,21 +247,36 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Load configuration values from environment variables
|
||||
s := &Config{
|
||||
return &Config{
|
||||
DataDir: envString("DATA_DIR"),
|
||||
Debug: envBool("DEBUG", false),
|
||||
MaintenanceMode: envBool("MAINTENANCE_MODE", false),
|
||||
Debug: debug,
|
||||
MaintenanceMode: maintenanceMode,
|
||||
Environment: environment,
|
||||
MetricsUsername: envString("METRICS_USERNAME"),
|
||||
MetricsPassword: envString("METRICS_PASSWORD"),
|
||||
Port: envInt("PORT", defaultPort),
|
||||
Port: port,
|
||||
SentryDSN: envString("SENTRY_DSN"),
|
||||
RetentionSweepInterval: retentionSweepInterval,
|
||||
log: log,
|
||||
params: ¶ms,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// New creates a Config by reading environment variables.
|
||||
//
|
||||
//nolint:revive // lc parameter is required by fx even if unused.
|
||||
func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
|
||||
log := params.Logger.Get()
|
||||
|
||||
// A set-but-unparseable value anywhere in the environment is a
|
||||
// hard error, so fx aborts startup rather than running with a
|
||||
// silently substituted default.
|
||||
s, err := loadFromEnv()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.log = log
|
||||
s.params = ¶ms
|
||||
|
||||
// Set default DataDir. All SQLite databases (main application
|
||||
// DB and per-webhook event DBs) live here. The same default is
|
||||
// used regardless of environment; override with DATA_DIR if
|
||||
|
||||
409
internal/config/env_test.go
Normal file
409
internal/config/env_test.go
Normal file
@@ -0,0 +1,409 @@
|
||||
package config_test
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
"sneak.berlin/go/webhooker/internal/globals"
|
||||
"sneak.berlin/go/webhooker/internal/logger"
|
||||
)
|
||||
|
||||
// testEnvKey is a throwaway variable name used only by the helper
|
||||
// tables below, so they cannot disturb real configuration.
|
||||
const testEnvKey = "WEBHOOKER_TEST_VALUE"
|
||||
|
||||
// Real configuration variables exercised by the config.New tests.
|
||||
const (
|
||||
envKeyPort = "PORT"
|
||||
envKeyDebug = "DEBUG"
|
||||
envKeyMaintenanceMode = "MAINTENANCE_MODE"
|
||||
)
|
||||
|
||||
// envBoolCase is one row of the envBool table.
|
||||
type envBoolCase struct {
|
||||
name string
|
||||
set bool
|
||||
value string
|
||||
defaultValue bool
|
||||
expectError bool
|
||||
expected bool
|
||||
}
|
||||
|
||||
// envBoolCases is the envBool table, kept out of the test body so
|
||||
// the test itself stays readable.
|
||||
func envBoolCases() []envBoolCase {
|
||||
return []envBoolCase{
|
||||
{
|
||||
name: "unset uses default false",
|
||||
defaultValue: false,
|
||||
expected: false,
|
||||
},
|
||||
{
|
||||
name: "unset uses default true",
|
||||
defaultValue: true,
|
||||
expected: true,
|
||||
},
|
||||
{
|
||||
name: "empty uses default true",
|
||||
set: true,
|
||||
value: "",
|
||||
defaultValue: true,
|
||||
expected: true,
|
||||
},
|
||||
{
|
||||
name: "true is parsed",
|
||||
set: true,
|
||||
value: "true",
|
||||
expected: true,
|
||||
},
|
||||
{
|
||||
name: "one is parsed",
|
||||
set: true,
|
||||
value: "1",
|
||||
expected: true,
|
||||
},
|
||||
{
|
||||
name: "False is parsed",
|
||||
set: true,
|
||||
value: "False",
|
||||
defaultValue: true,
|
||||
expected: false,
|
||||
},
|
||||
{
|
||||
name: "zero is parsed",
|
||||
set: true,
|
||||
value: "0",
|
||||
defaultValue: true,
|
||||
expected: false,
|
||||
},
|
||||
{
|
||||
name: "yes is rejected",
|
||||
set: true,
|
||||
value: "yes",
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "on is rejected",
|
||||
set: true,
|
||||
value: "on",
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "typo is rejected",
|
||||
set: true,
|
||||
value: "ture",
|
||||
expectError: true,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnvBool(t *testing.T) {
|
||||
for _, tt := range envBoolCases() {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Cannot use t.Parallel() here because t.Setenv
|
||||
// is incompatible with parallel subtests.
|
||||
if tt.set {
|
||||
t.Setenv(testEnvKey, tt.value)
|
||||
} else {
|
||||
require.NoError(t, os.Unsetenv(testEnvKey))
|
||||
}
|
||||
|
||||
got, err := config.EnvBoolForTest(
|
||||
testEnvKey, tt.defaultValue,
|
||||
)
|
||||
|
||||
if tt.expectError {
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), testEnvKey)
|
||||
assert.Contains(t, err.Error(), tt.value)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tt.expected, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnvPositiveInt(t *testing.T) {
|
||||
const defaultValue = 7
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
set bool
|
||||
value string
|
||||
expectError bool
|
||||
errIs error
|
||||
expected int
|
||||
}{
|
||||
{
|
||||
name: "unset returns the default integer",
|
||||
expected: defaultValue,
|
||||
},
|
||||
{
|
||||
name: "empty returns the default integer",
|
||||
set: true,
|
||||
value: "",
|
||||
expected: defaultValue,
|
||||
},
|
||||
{
|
||||
name: "positive value is parsed",
|
||||
set: true,
|
||||
value: "42",
|
||||
expected: 42,
|
||||
},
|
||||
{
|
||||
name: "unparseable value is rejected",
|
||||
set: true,
|
||||
value: "not-a-number",
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "zero is rejected",
|
||||
set: true,
|
||||
value: "0",
|
||||
expectError: true,
|
||||
errIs: config.ErrNonPositiveValue,
|
||||
},
|
||||
{
|
||||
name: "negative is rejected",
|
||||
set: true,
|
||||
value: "-5",
|
||||
expectError: true,
|
||||
errIs: config.ErrNonPositiveValue,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Cannot use t.Parallel() here because t.Setenv
|
||||
// is incompatible with parallel subtests.
|
||||
if tt.set {
|
||||
t.Setenv(testEnvKey, tt.value)
|
||||
} else {
|
||||
require.NoError(t, os.Unsetenv(testEnvKey))
|
||||
}
|
||||
|
||||
got, err := config.EnvPositiveIntForTest(
|
||||
testEnvKey, defaultValue,
|
||||
)
|
||||
|
||||
if tt.expectError {
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), testEnvKey)
|
||||
assert.Contains(t, err.Error(), tt.value)
|
||||
|
||||
if tt.errIs != nil {
|
||||
require.ErrorIs(t, err, tt.errIs)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tt.expected, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnvPort(t *testing.T) {
|
||||
const defaultValue = 8080
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
set bool
|
||||
value string
|
||||
expectError bool
|
||||
errIs error
|
||||
expected int
|
||||
}{
|
||||
{
|
||||
name: "unset returns the default port",
|
||||
expected: defaultValue,
|
||||
},
|
||||
{
|
||||
name: "valid port is parsed",
|
||||
set: true,
|
||||
value: "9000",
|
||||
expected: 9000,
|
||||
},
|
||||
{
|
||||
name: "highest port is accepted",
|
||||
set: true,
|
||||
value: "65535",
|
||||
expected: 65535,
|
||||
},
|
||||
{
|
||||
name: "unparseable value is rejected",
|
||||
set: true,
|
||||
value: "not-a-port",
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "zero is rejected",
|
||||
set: true,
|
||||
value: "0",
|
||||
expectError: true,
|
||||
errIs: config.ErrNonPositiveValue,
|
||||
},
|
||||
{
|
||||
name: "above the port range is rejected",
|
||||
set: true,
|
||||
value: "65536",
|
||||
expectError: true,
|
||||
errIs: config.ErrInvalidPort,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Cannot use t.Parallel() here because t.Setenv
|
||||
// is incompatible with parallel subtests.
|
||||
if tt.set {
|
||||
t.Setenv(testEnvKey, tt.value)
|
||||
} else {
|
||||
require.NoError(t, os.Unsetenv(testEnvKey))
|
||||
}
|
||||
|
||||
got, err := config.EnvPortForTest(
|
||||
testEnvKey, defaultValue,
|
||||
)
|
||||
|
||||
if tt.expectError {
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), testEnvKey)
|
||||
|
||||
if tt.errIs != nil {
|
||||
require.ErrorIs(t, err, tt.errIs)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tt.expected, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// buildConfig constructs a Config through fx exactly as the
|
||||
// application does, returning the config and any construction error.
|
||||
func buildConfig(t *testing.T) (*config.Config, error) {
|
||||
t.Helper()
|
||||
|
||||
var cfg *config.Config
|
||||
|
||||
app := fx.New(
|
||||
fx.NopLogger,
|
||||
fx.Provide(
|
||||
globals.New,
|
||||
logger.New,
|
||||
config.New,
|
||||
),
|
||||
fx.Populate(&cfg),
|
||||
)
|
||||
|
||||
return cfg, app.Err()
|
||||
}
|
||||
|
||||
func TestNewRejectsBadEnvValues(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
key string
|
||||
value string
|
||||
expectError bool
|
||||
check func(t *testing.T, cfg *config.Config)
|
||||
}{
|
||||
{
|
||||
name: "valid PORT is used",
|
||||
key: envKeyPort,
|
||||
value: "9001",
|
||||
check: func(t *testing.T, cfg *config.Config) {
|
||||
t.Helper()
|
||||
assert.Equal(t, 9001, cfg.Port)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "unparseable PORT aborts startup",
|
||||
key: envKeyPort,
|
||||
value: "eighty-eighty",
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "out-of-range PORT aborts startup",
|
||||
key: envKeyPort,
|
||||
value: "70000",
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "valid DEBUG is used",
|
||||
key: envKeyDebug,
|
||||
value: "true",
|
||||
check: func(t *testing.T, cfg *config.Config) {
|
||||
t.Helper()
|
||||
assert.True(t, cfg.Debug)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "unparseable DEBUG aborts startup",
|
||||
key: envKeyDebug,
|
||||
value: "ture",
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "unparseable MAINTENANCE_MODE aborts startup",
|
||||
key: envKeyMaintenanceMode,
|
||||
value: "sometimes",
|
||||
expectError: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Cannot use t.Parallel() here because t.Setenv
|
||||
// is incompatible with parallel subtests.
|
||||
t.Setenv("WEBHOOKER_ENVIRONMENT", "dev")
|
||||
t.Setenv(tt.key, tt.value)
|
||||
|
||||
cfg, err := buildConfig(t)
|
||||
|
||||
if tt.expectError {
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), tt.key)
|
||||
assert.Contains(t, err.Error(), tt.value)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, cfg)
|
||||
tt.check(t, cfg)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestNewUsesDefaultsWhenUnset proves the fail-loud behaviour did not
|
||||
// break the legitimate unset case: absent variables still get their
|
||||
// documented defaults.
|
||||
func TestNewUsesDefaultsWhenUnset(t *testing.T) {
|
||||
t.Setenv("WEBHOOKER_ENVIRONMENT", "dev")
|
||||
|
||||
for _, key := range []string{
|
||||
envKeyPort, envKeyDebug, envKeyMaintenanceMode,
|
||||
} {
|
||||
require.NoError(t, os.Unsetenv(key))
|
||||
}
|
||||
|
||||
cfg, err := buildConfig(t)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, cfg)
|
||||
|
||||
assert.Equal(t, 8080, cfg.Port)
|
||||
assert.False(t, cfg.Debug)
|
||||
assert.False(t, cfg.MaintenanceMode)
|
||||
}
|
||||
20
internal/config/export_test.go
Normal file
20
internal/config/export_test.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package config
|
||||
|
||||
// This file exposes the unexported environment parsing helpers to
|
||||
// the external config_test package so each helper can be covered by
|
||||
// its own table-driven test without weakening the package API.
|
||||
|
||||
// EnvBoolForTest exposes envBool.
|
||||
func EnvBoolForTest(key string, defaultValue bool) (bool, error) {
|
||||
return envBool(key, defaultValue)
|
||||
}
|
||||
|
||||
// EnvPositiveIntForTest exposes envPositiveInt.
|
||||
func EnvPositiveIntForTest(key string, defaultValue int) (int, error) {
|
||||
return envPositiveInt(key, defaultValue)
|
||||
}
|
||||
|
||||
// EnvPortForTest exposes envPort.
|
||||
func EnvPortForTest(key string, defaultValue int) (int, error) {
|
||||
return envPort(key, defaultValue)
|
||||
}
|
||||
@@ -5,8 +5,6 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"go.uber.org/fx"
|
||||
)
|
||||
|
||||
// NewTestRetentionReaper builds a RetentionReaper backed by the given
|
||||
@@ -31,26 +29,3 @@ func NewTestRetentionReaper(
|
||||
func (r *RetentionReaper) ExportSweep(ctx context.Context) {
|
||||
r.sweep(ctx)
|
||||
}
|
||||
|
||||
// ExportRegisterHooks registers the reaper's real fx lifecycle hooks
|
||||
// on a lifecycle supplied by a test, so a test can drive the exact
|
||||
// OnStart/OnStop functions the application runs and hand OnStart the
|
||||
// kind of context fx actually supplies.
|
||||
func (r *RetentionReaper) ExportRegisterHooks(lc fx.Lifecycle) {
|
||||
r.registerHooks(lc)
|
||||
}
|
||||
|
||||
// ExportStart starts the reaper's background loop for tests.
|
||||
func (r *RetentionReaper) ExportStart() {
|
||||
r.start()
|
||||
}
|
||||
|
||||
// ExportStop stops the reaper's background loop for tests.
|
||||
func (r *RetentionReaper) ExportStop() {
|
||||
r.stop()
|
||||
}
|
||||
|
||||
// ExportSetInterval overrides the sweep interval for tests.
|
||||
func (r *RetentionReaper) ExportSetInterval(d time.Duration) {
|
||||
r.interval = d
|
||||
}
|
||||
|
||||
@@ -56,20 +56,9 @@ func NewRetentionReaper(
|
||||
interval: params.Config.RetentionSweepInterval,
|
||||
}
|
||||
|
||||
r.registerHooks(lc)
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// registerHooks wires the reaper's start and stop into the fx
|
||||
// lifecycle. The start hook's context is deliberately ignored: see
|
||||
// start for why the sweep loop must not inherit it.
|
||||
func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
|
||||
lc.Append(fx.Hook{
|
||||
//nolint:contextcheck // Not inheriting the hook context is
|
||||
// the point: see start.
|
||||
OnStart: func(_ context.Context) error {
|
||||
r.start()
|
||||
OnStart: func(ctx context.Context) error {
|
||||
r.start(ctx)
|
||||
|
||||
return nil
|
||||
},
|
||||
@@ -79,20 +68,12 @@ func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
|
||||
return nil
|
||||
},
|
||||
})
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// start launches the background sweep loop.
|
||||
//
|
||||
// The loop's context is derived from context.Background(), NOT from
|
||||
// the fx OnStart hook context. The hook context carries fx's start
|
||||
// timeout (15s by default) and is cancelled once the start phase
|
||||
// completes, so a loop derived from it dies 45 minutes before its
|
||||
// first tick under the default one-hour sweep interval, leaving a
|
||||
// reaper that never reaps. A long-lived goroutine must outlive the
|
||||
// startup phase, so its lifetime is bounded by OnStop instead: stop
|
||||
// cancels this context and waits on the WaitGroup.
|
||||
func (r *RetentionReaper) start() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
func (r *RetentionReaper) start(ctx context.Context) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
r.cancel = cancel
|
||||
|
||||
r.wg.Add(1)
|
||||
|
||||
@@ -1,209 +0,0 @@
|
||||
package database_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/fx"
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
)
|
||||
|
||||
const (
|
||||
// reaperTestInterval is the sweep interval a lifecycle test
|
||||
// runs the reaper at, so a loop that survives startup produces
|
||||
// an observable sweep quickly.
|
||||
reaperTestInterval = 10 * time.Millisecond
|
||||
|
||||
// reaperStopTimeout bounds how long a lifecycle test waits for
|
||||
// the reaper's OnStop hook to return before declaring the
|
||||
// shutdown hung.
|
||||
reaperStopTimeout = 10 * time.Second
|
||||
|
||||
// reaperTestRetentionDays is the retention policy the lifecycle
|
||||
// tests give their webhook.
|
||||
reaperTestRetentionDays = 30
|
||||
)
|
||||
|
||||
// recordingLifecycle is a minimal fx.Lifecycle that records the
|
||||
// hooks a component registers, so a test can invoke the real
|
||||
// OnStart/OnStop functions with a context of its choosing.
|
||||
type recordingLifecycle struct {
|
||||
hooks []fx.Hook
|
||||
}
|
||||
|
||||
func (l *recordingLifecycle) Append(h fx.Hook) {
|
||||
l.hooks = append(l.hooks, h)
|
||||
}
|
||||
|
||||
// startReaperViaHook drives the genuine fx hooks the application
|
||||
// registers for the reaper, handing OnStart a context that is
|
||||
// already done. It returns the recorded lifecycle so the caller
|
||||
// can drive OnStop too.
|
||||
func startReaperViaHook(
|
||||
t *testing.T, r *database.RetentionReaper,
|
||||
) *recordingLifecycle {
|
||||
t.Helper()
|
||||
|
||||
lc := &recordingLifecycle{}
|
||||
r.ExportRegisterHooks(lc)
|
||||
require.Len(t, lc.hooks, 1)
|
||||
|
||||
// fx hands OnStart a context carrying the application start
|
||||
// timeout, and cancels it when the start phase ends. An
|
||||
// already-cancelled context is that same defect taken to its
|
||||
// limit, and unlike a plain context.Background() it actually
|
||||
// distinguishes a correctly rooted loop from a broken one.
|
||||
hookCtx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
require.NoError(t, lc.hooks[0].OnStart(hookCtx))
|
||||
|
||||
return lc
|
||||
}
|
||||
|
||||
// eventGone reports whether an event row has been removed. It
|
||||
// takes no *testing.T because it is polled from an
|
||||
// assert.Eventually condition, which runs off the test goroutine
|
||||
// where testify assertions must not be used.
|
||||
func eventGone(db *gorm.DB, eventID string) bool {
|
||||
var n int64
|
||||
|
||||
err := db.Unscoped().Model(&database.Event{}).
|
||||
Where("id = ?", eventID).Count(&n).Error
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
return n == 0
|
||||
}
|
||||
|
||||
// seedExpiredWebhook creates a webhook with a finite retention
|
||||
// policy plus one long-expired event chain, and returns the
|
||||
// webhook's database and the chain's event ID.
|
||||
func seedExpiredWebhook(
|
||||
t *testing.T, env *retentionTestEnv,
|
||||
) (*gorm.DB, string) {
|
||||
t.Helper()
|
||||
|
||||
webhookID := createWebhook(
|
||||
t, env.mainDB.DB(), reaperTestRetentionDays,
|
||||
)
|
||||
|
||||
db, err := env.mgr.GetDB(webhookID)
|
||||
require.NoError(t, err)
|
||||
|
||||
chain := seedEventChain(
|
||||
t, db, webhookID,
|
||||
time.Now().Add(-365*24*time.Hour),
|
||||
)
|
||||
|
||||
return db, chain.eventID
|
||||
}
|
||||
|
||||
// TestRetentionReaper_LoopOutlivesStartHookContext is the
|
||||
// regression test for a reaper that never reaped. fx calls
|
||||
// OnStart with a context carrying the application's start timeout
|
||||
// (15s by default) and cancels it when the start phase ends, so a
|
||||
// sweep loop rooted in it is dead three quarters of an hour
|
||||
// before its first tick under the default one-hour interval, and
|
||||
// per-webhook event databases grow without bound exactly as they
|
||||
// did before retention existed.
|
||||
//
|
||||
// Driving OnStart with an already-cancelled context is that
|
||||
// defect taken to its limit: a loop that inherits the hook
|
||||
// context never ticks once, while a correctly rooted loop keeps
|
||||
// sweeping for as long as the process lives.
|
||||
func TestRetentionReaper_LoopOutlivesStartHookContext(
|
||||
t *testing.T,
|
||||
) {
|
||||
t.Parallel()
|
||||
|
||||
env := setupRetentionTest(t)
|
||||
|
||||
db, eventID := seedExpiredWebhook(t, env)
|
||||
|
||||
env.reaper.ExportSetInterval(reaperTestInterval)
|
||||
|
||||
lc := startReaperViaHook(t, env.reaper)
|
||||
t.Cleanup(func() {
|
||||
_ = lc.hooks[0].OnStop(context.Background())
|
||||
})
|
||||
|
||||
assert.Eventually(
|
||||
t,
|
||||
func() bool { return eventGone(db, eventID) },
|
||||
5*time.Second,
|
||||
reaperTestInterval,
|
||||
"the sweep loop must keep running after the start "+
|
||||
"hook's context is done; it reaped nothing, so it "+
|
||||
"inherited the hook context and died",
|
||||
)
|
||||
}
|
||||
|
||||
// TestRetentionReaper_StopHookStopsLoop proves the fix did not
|
||||
// trade a startup bug for a shutdown hang: now that the sweep
|
||||
// loop no longer observes the start hook's cancellation, OnStop
|
||||
// is the only thing that can stop it, and it must both return
|
||||
// promptly and actually leave the loop stopped.
|
||||
func TestRetentionReaper_StopHookStopsLoop(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
env := setupRetentionTest(t)
|
||||
|
||||
db, eventID := seedExpiredWebhook(t, env)
|
||||
|
||||
env.reaper.ExportSetInterval(reaperTestInterval)
|
||||
|
||||
lc := startReaperViaHook(t, env.reaper)
|
||||
|
||||
// Let the loop prove it is running before stopping it, so a
|
||||
// fast OnStop cannot pass by stopping something already dead.
|
||||
require.Eventually(
|
||||
t,
|
||||
func() bool { return eventGone(db, eventID) },
|
||||
5*time.Second,
|
||||
reaperTestInterval,
|
||||
)
|
||||
|
||||
var stopErr error
|
||||
|
||||
stopped := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
defer close(stopped)
|
||||
|
||||
// stop blocks on the loop's WaitGroup, so returning at all
|
||||
// proves the goroutine observed the cancellation.
|
||||
stopErr = lc.hooks[0].OnStop(context.Background())
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-stopped:
|
||||
case <-time.After(reaperStopTimeout):
|
||||
t.Fatal(
|
||||
"OnStop did not return: the retention reaper's " +
|
||||
"WaitGroup is still waiting on a loop that never " +
|
||||
"observed cancellation",
|
||||
)
|
||||
}
|
||||
|
||||
require.NoError(t, stopErr)
|
||||
|
||||
// With the loop gone, a newly expired chain must survive.
|
||||
survivor := seedEventChain(
|
||||
t, db, "stopped-webhook",
|
||||
time.Now().Add(-365*24*time.Hour),
|
||||
)
|
||||
|
||||
time.Sleep(20 * reaperTestInterval)
|
||||
|
||||
assert.False(
|
||||
t,
|
||||
eventGone(db, survivor.eventID),
|
||||
"a stopped reaper must not sweep anything",
|
||||
)
|
||||
}
|
||||
@@ -149,7 +149,18 @@ func New(
|
||||
Transport: NewSSRFSafeTransport(),
|
||||
})
|
||||
|
||||
e.registerHooks(lc)
|
||||
lc.Append(fx.Hook{
|
||||
OnStart: func(ctx context.Context) error {
|
||||
e.start(ctx)
|
||||
|
||||
return nil
|
||||
},
|
||||
OnStop: func(_ context.Context) error {
|
||||
e.stop()
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
|
||||
return e
|
||||
}
|
||||
@@ -199,40 +210,8 @@ func (e *Engine) ScheduleRetry(
|
||||
})
|
||||
}
|
||||
|
||||
// registerHooks wires the engine's start and stop into the fx
|
||||
// lifecycle. The start hook's context is deliberately ignored:
|
||||
// see start for why the worker pool must not inherit it.
|
||||
func (e *Engine) registerHooks(lc fx.Lifecycle) {
|
||||
lc.Append(fx.Hook{
|
||||
//nolint:contextcheck // Not inheriting the hook context
|
||||
// is the point: see start.
|
||||
OnStart: func(_ context.Context) error {
|
||||
e.start()
|
||||
|
||||
return nil
|
||||
},
|
||||
OnStop: func(_ context.Context) error {
|
||||
e.stop()
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// start launches the worker pool, restart recovery, and the
|
||||
// periodic retry sweep.
|
||||
//
|
||||
// Their context is derived from context.Background(), NOT from
|
||||
// the fx OnStart hook context. The hook context carries fx's
|
||||
// start timeout (15s by default) and is cancelled once the start
|
||||
// phase completes, so goroutines derived from it stop a few
|
||||
// seconds into the process: every worker would return and the
|
||||
// engine would silently stop delivering webhooks entirely. A
|
||||
// long-lived goroutine must outlive the startup phase, so its
|
||||
// lifetime is bounded by OnStop instead: stop cancels this
|
||||
// context and waits on the WaitGroup.
|
||||
func (e *Engine) start() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
func (e *Engine) start(ctx context.Context) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
e.cancel = cancel
|
||||
|
||||
for range e.workers {
|
||||
|
||||
@@ -476,7 +476,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
s.Engine.ExportStart()
|
||||
s.Engine.ExportStart(context.Background())
|
||||
|
||||
event := iSeedEvent(
|
||||
t, s.WebhookDB, s.WebhookID,
|
||||
@@ -499,17 +499,21 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
|
||||
|
||||
s.Engine.Notify([]delivery.Task{task})
|
||||
|
||||
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
||||
iWaitForStatus(
|
||||
t, s.WebhookDB, d.ID,
|
||||
database.DeliveryStatusDelivered,
|
||||
)
|
||||
|
||||
s.Engine.ExportStop()
|
||||
}
|
||||
|
||||
// iWaitForDelivered polls until the delivery reaches the
|
||||
// delivered status.
|
||||
func iWaitForDelivered(
|
||||
// iWaitForStatus polls until the delivery reaches the
|
||||
// expected status.
|
||||
func iWaitForStatus(
|
||||
t *testing.T,
|
||||
db *gorm.DB,
|
||||
deliveryID string,
|
||||
expected database.DeliveryStatus,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
@@ -523,7 +527,7 @@ func iWaitForDelivered(
|
||||
return false
|
||||
}
|
||||
|
||||
return d.Status == database.DeliveryStatusDelivered
|
||||
return d.Status == expected
|
||||
}, 5*time.Second, 50*time.Millisecond)
|
||||
}
|
||||
|
||||
@@ -554,7 +558,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
|
||||
database.DeliveryStatusRetrying,
|
||||
)
|
||||
|
||||
s.Engine.ExportStart()
|
||||
s.Engine.ExportStart(context.Background())
|
||||
|
||||
bodyStr := event.Body
|
||||
cfg := iHTTPConfig(ts.URL)
|
||||
@@ -565,7 +569,10 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
|
||||
|
||||
s.Engine.ExportRetryCh() <- task
|
||||
|
||||
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
||||
iWaitForStatus(
|
||||
t, s.WebhookDB, d.ID,
|
||||
database.DeliveryStatusDelivered,
|
||||
)
|
||||
|
||||
s.Engine.ExportStop()
|
||||
}
|
||||
|
||||
@@ -1,199 +0,0 @@
|
||||
package delivery_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/delivery"
|
||||
)
|
||||
|
||||
const (
|
||||
// hookStopTimeout bounds how long a lifecycle test waits for
|
||||
// the engine's OnStop hook to return before declaring the
|
||||
// shutdown hung.
|
||||
hookStopTimeout = 10 * time.Second
|
||||
|
||||
// hookSettleDelay is how long startEngineViaHook waits after
|
||||
// OnStart before the caller may enqueue work. A worker pool
|
||||
// wrongly rooted in the already-done hook context has nothing
|
||||
// but ctx.Done() ready in its select, so it is deterministically
|
||||
// gone by the end of this window. Without the wait, Notify would
|
||||
// race the pool's very first select, in which a ready ctx.Done()
|
||||
// and a ready deliveryCh are chosen between at random and a
|
||||
// doomed pool still delivers.
|
||||
hookSettleDelay = 250 * time.Millisecond
|
||||
)
|
||||
|
||||
// recordingLifecycle is a minimal fx.Lifecycle that records the
|
||||
// hooks a component registers, so a test can invoke the real
|
||||
// OnStart/OnStop functions with a context of its choosing.
|
||||
type recordingLifecycle struct {
|
||||
hooks []fx.Hook
|
||||
}
|
||||
|
||||
func (l *recordingLifecycle) Append(h fx.Hook) {
|
||||
l.hooks = append(l.hooks, h)
|
||||
}
|
||||
|
||||
// startEngineViaHook drives the genuine fx hooks the application
|
||||
// registers for the engine, handing OnStart a context that is
|
||||
// already done, and returns only once a pool that inherited that
|
||||
// context would have exited. It returns the recorded lifecycle so
|
||||
// the caller can drive OnStop too.
|
||||
//
|
||||
// Callers must not seed pending or retrying deliveries before
|
||||
// calling this: restart recovery enqueues those during startup,
|
||||
// which would put work in the queue while the pool is still
|
||||
// racing its first select.
|
||||
func startEngineViaHook(
|
||||
t *testing.T, eng *delivery.Engine,
|
||||
) *recordingLifecycle {
|
||||
t.Helper()
|
||||
|
||||
lc := &recordingLifecycle{}
|
||||
eng.ExportRegisterHooks(lc)
|
||||
require.Len(t, lc.hooks, 1)
|
||||
|
||||
// fx hands OnStart a context carrying the application start
|
||||
// timeout, and cancels it when the start phase ends. An
|
||||
// already-cancelled context is that same defect taken to its
|
||||
// limit, and unlike a plain context.Background() it actually
|
||||
// distinguishes a correctly rooted loop from a broken one.
|
||||
hookCtx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
require.NoError(t, lc.hooks[0].OnStart(hookCtx))
|
||||
|
||||
time.Sleep(hookSettleDelay)
|
||||
|
||||
return lc
|
||||
}
|
||||
|
||||
// seedLogTask seeds a pending delivery for a log target and
|
||||
// returns its ID together with the task that drives it. The log
|
||||
// target needs no network, so a delivery completing proves only
|
||||
// that a worker picked the task up.
|
||||
func seedLogTask(
|
||||
t *testing.T, s iSetup,
|
||||
) (string, delivery.Task) {
|
||||
t.Helper()
|
||||
|
||||
event := iSeedEvent(
|
||||
t, s.WebhookDB, s.WebhookID,
|
||||
`{"lifecycle":"hook-context"}`,
|
||||
)
|
||||
targetID := uuid.New().String()
|
||||
|
||||
d := iSeedDelivery(
|
||||
t, s.WebhookDB, event.ID, targetID,
|
||||
database.DeliveryStatusPending,
|
||||
)
|
||||
|
||||
bodyStr := event.Body
|
||||
task := iTask(
|
||||
d, event, s.WebhookID, targetID,
|
||||
"hook-context-test", "", 0, 1, &bodyStr,
|
||||
)
|
||||
task.TargetType = database.TargetTypeLog
|
||||
|
||||
return d.ID, task
|
||||
}
|
||||
|
||||
// TestEngine_WorkersOutliveStartHookContext is the regression
|
||||
// test for a delivery engine that stopped delivering roughly
|
||||
// fifteen seconds after boot. fx calls OnStart with a context
|
||||
// carrying the application's start timeout (15s by default) and
|
||||
// cancels it when the start phase ends, so a worker pool rooted
|
||||
// in it exits shortly after startup: the process keeps accepting
|
||||
// and persisting events while nothing at all forwards them.
|
||||
//
|
||||
// Driving OnStart with an already-cancelled context is that
|
||||
// defect taken to its limit. A pool that inherits the hook
|
||||
// context is gone before the task is even enqueued; a correctly
|
||||
// rooted pool keeps working for as long as the process lives.
|
||||
func TestEngine_WorkersOutliveStartHookContext(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
|
||||
lc := startEngineViaHook(t, s.Engine)
|
||||
t.Cleanup(func() {
|
||||
_ = lc.hooks[0].OnStop(context.Background())
|
||||
})
|
||||
|
||||
// Seeded only after the pool has settled, so restart recovery
|
||||
// cannot enqueue it during startup.
|
||||
deliveryID, task := seedLogTask(t, s)
|
||||
|
||||
s.Engine.Notify([]delivery.Task{task})
|
||||
|
||||
iWaitForDelivered(t, s.WebhookDB, deliveryID)
|
||||
}
|
||||
|
||||
// TestEngine_StopHookStopsWorkers proves the fix did not trade a
|
||||
// startup bug for a shutdown hang: now that the worker pool no
|
||||
// longer observes the start hook's cancellation, OnStop is the
|
||||
// only thing that can stop it, and it must both return promptly
|
||||
// and actually leave the pool drained.
|
||||
func TestEngine_StopHookStopsWorkers(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
|
||||
lc := startEngineViaHook(t, s.Engine)
|
||||
|
||||
// Let the pool prove it is running before stopping it, so a
|
||||
// fast OnStop cannot pass by stopping something already dead.
|
||||
firstID, firstTask := seedLogTask(t, s)
|
||||
s.Engine.Notify([]delivery.Task{firstTask})
|
||||
iWaitForDelivered(t, s.WebhookDB, firstID)
|
||||
|
||||
var stopErr error
|
||||
|
||||
stopped := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
defer close(stopped)
|
||||
|
||||
// stop blocks on the workers' WaitGroup, so returning at
|
||||
// all proves every goroutine observed the cancellation.
|
||||
stopErr = lc.hooks[0].OnStop(context.Background())
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-stopped:
|
||||
case <-time.After(hookStopTimeout):
|
||||
t.Fatal(
|
||||
"OnStop did not return: the delivery engine's " +
|
||||
"WaitGroup is still waiting on a goroutine that " +
|
||||
"never observed cancellation",
|
||||
)
|
||||
}
|
||||
|
||||
require.NoError(t, stopErr)
|
||||
|
||||
// With every worker gone, a freshly notified task must sit
|
||||
// untouched in the queue rather than being delivered.
|
||||
secondID, secondTask := seedLogTask(t, s)
|
||||
s.Engine.Notify([]delivery.Task{secondTask})
|
||||
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
|
||||
var after database.Delivery
|
||||
|
||||
require.NoError(
|
||||
t,
|
||||
s.WebhookDB.First(&after, "id = ?", secondID).Error,
|
||||
)
|
||||
require.Equal(
|
||||
t,
|
||||
database.DeliveryStatusPending,
|
||||
after.Status,
|
||||
"a stopped engine must not deliver anything",
|
||||
)
|
||||
}
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"go.uber.org/fx"
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
)
|
||||
@@ -190,16 +189,8 @@ func (e *Engine) ExportRecoverInFlight(
|
||||
}
|
||||
|
||||
// ExportStart exposes start for testing.
|
||||
func (e *Engine) ExportStart() {
|
||||
e.start()
|
||||
}
|
||||
|
||||
// ExportRegisterHooks registers the engine's real fx lifecycle
|
||||
// hooks on a lifecycle supplied by a test, so a test can drive
|
||||
// the exact OnStart/OnStop functions the application runs and
|
||||
// hand OnStart the kind of context fx actually supplies.
|
||||
func (e *Engine) ExportRegisterHooks(lc fx.Lifecycle) {
|
||||
e.registerHooks(lc)
|
||||
func (e *Engine) ExportStart(ctx context.Context) {
|
||||
e.start(ctx)
|
||||
}
|
||||
|
||||
// ExportStop exposes stop for testing.
|
||||
|
||||
Reference in New Issue
Block a user