Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sneak
|
a9e2e07217 | ||
|
sneak
|
54b643044f | ||
 clawbot
|
55ccb28270 | ||
|
clawbot
|
06be075b70 | ||
|
sneak
|
766f33a956 |
+13
-8
@@ -111,11 +111,16 @@ RUN --network=none golangci-lint run --config .golangci.yml --build-tags browser
|
||||
# Nothing is wanted from js-lint; the copy is what makes this phase run it.
|
||||
COPY --from=js-lint /src/yarn.lock /dev/null
|
||||
|
||||
# Test phase. -race needs cgo and so a C compiler, which the Debian Go image
|
||||
# ships and the alpine one does not. `make test` (script/test) builds this
|
||||
# stage alone; the build stage below depends on it.
|
||||
# Test phase. The tests run with cgo off and so without Go's race detector,
|
||||
# which needs cgo on Linux; sneak ruled the race detector out in
|
||||
# https://git.eeqj.de/sneak/webhooker/issues/538#issuecomment-153273.
|
||||
# `make test` (script/test) builds this stage alone; the build stage below
|
||||
# depends on it.
|
||||
#
|
||||
# golang:1.26.1-bookworm (Debian-based), 2026-03-17
|
||||
# golang:1.26.1-bookworm (Debian-based), 2026-03-17: the builder stage's
|
||||
# image, which that stage needs for its git and make. Its test step runs
|
||||
# under the image's bash, and the tests in internal/versionscript use its
|
||||
# git, make and sh, and skip when they are missing.
|
||||
FROM golang:1.26.1-bookworm@sha256:4465644228bc2857a954b092167e12aa59c006a3492282a6c820bf4755fd64a4 AS test
|
||||
|
||||
WORKDIR /src
|
||||
@@ -131,8 +136,8 @@ RUN script/assets
|
||||
# -timeout applies to each package on its own, so 90s has only to clear the
|
||||
# slowest one. -p 4 -parallel 8 keep the run under 2 GB of memory: at most
|
||||
# four test binaries build or run at once, each with at most eight parallel
|
||||
# tests. Under -race every test binary and every link costs a few hundred MB,
|
||||
# so the defaults (one per core) add up to several GB on a many-core host.
|
||||
# tests. The defaults are one per core, so without these the memory the run
|
||||
# needs grows with the host's core count.
|
||||
#
|
||||
# The first run has no -v: go test then prints one result line per package,
|
||||
# with its coverage, and for a package that fails, everything its tests
|
||||
@@ -154,12 +159,12 @@ RUN script/assets
|
||||
SHELL ["/bin/bash", "-o", "pipefail", "-c"]
|
||||
RUN --mount=type=tmpfs,target=/tmp/tests,size=512m \
|
||||
export TMPDIR=/tmp/tests GOTMPDIR=/tmp; \
|
||||
go test -race -cover -p 4 -parallel 8 -timeout 90s ./... 2>&1 | tee /tmp/go-test.log && exit 0; \
|
||||
CGO_ENABLED=0 go test -cover -p 4 -parallel 8 -timeout 90s ./... 2>&1 | tee /tmp/go-test.log && exit 0; \
|
||||
tests="$(awk '/^--- FAIL: / { print $3 }' /tmp/go-test.log | paste -s -d '|' -)"; \
|
||||
packages="$(awk '/^FAIL\t/ { print $2 }' /tmp/go-test.log)"; \
|
||||
if [ -n "$tests" ]; then \
|
||||
echo "--- Rerunning the failed tests with -v for details ---"; \
|
||||
go test -race -v -p 4 -parallel 8 -timeout 90s -run "^($tests)\$" $packages; \
|
||||
CGO_ENABLED=0 go test -v -p 4 -parallel 8 -timeout 90s -run "^($tests)\$" $packages; \
|
||||
fi; \
|
||||
exit 1
|
||||
|
||||
|
||||
+3
-3
@@ -15,9 +15,9 @@ RUN go mod download
|
||||
COPY . .
|
||||
|
||||
# The test binary embeds the templates and static files, so the browser
|
||||
# stage needs nothing else. -p 4 keeps the compile's memory down, as in
|
||||
# the test phase of Dockerfile.
|
||||
RUN make assets && go test -c -p 4 -tags browser -o /browser.test ./internal/server
|
||||
# stage needs nothing else. It is built with cgo off and -p 4 keeps the
|
||||
# compile's memory down, both as in the test phase of Dockerfile.
|
||||
RUN make assets && CGO_ENABLED=0 go test -c -p 4 -tags browser -o /browser.test ./internal/server
|
||||
|
||||
# chromedp/headless-shell:151.0.7922.109 (Debian trixie), 2026-08-11. The
|
||||
# browser is on PATH as headless-shell, where the test's browser library
|
||||
|
||||
@@ -60,7 +60,7 @@ make assets # Extract Alpine.js from 3p/ (build and dev run it)
|
||||
make fmt # Format Go (gofmt + goimports) and Markdown (prettier, in Docker)
|
||||
make fmt-check # Fail if gofmt or prettier would change anything (writes nothing)
|
||||
make lint # Run golangci-lint and ESLint in Docker
|
||||
make test # Run tests with race detection, in Docker
|
||||
make test # Run tests with cgo off, in Docker
|
||||
make test-browser # Run the browser test in Docker (Dockerfile.browser)
|
||||
make check # test + lint + fmt-check + css-check (CI gate)
|
||||
make build # Build binary to bin/webhooker (version-stamped)
|
||||
@@ -75,6 +75,11 @@ make css-check # Fail if static/css/tailwind.css is stale (writes nothing)
|
||||
make clean # Remove bin/
|
||||
```
|
||||
|
||||
The tests run with cgo off and so without Go's race detector, which needs cgo on
|
||||
Linux: an exception to the `-race` in the test phase of the vendored
|
||||
`REPO_POLICIES.md`, ruled by @sneak in
|
||||
https://git.eeqj.de/sneak/webhooker/issues/538#issuecomment-153273.
|
||||
|
||||
### Configuration
|
||||
|
||||
All configuration is via environment variables. For local development, you can
|
||||
@@ -2153,12 +2158,8 @@ External Service
|
||||
1. Look up Entrypoint by UUID — 404 if unknown,
|
||||
410 if inactive
|
||||
2. Read the body under the 1 MB cap
|
||||
3. Capture full request as Event
|
||||
4. Create Delivery records for each active Target
|
||||
5. Build self-contained delivery.Task structs
|
||||
(target settings + event data inline for
|
||||
bodies < 16 KiB)
|
||||
6. Notify Engine via channel (no DB read needed)
|
||||
3. Hand the full request, as an Event, to
|
||||
Engine.StoreEvent
|
||||
│
|
||||
▼
|
||||
┌──────────────┐
|
||||
@@ -2167,6 +2168,13 @@ External Service
|
||||
│ (worker │
|
||||
│ pool) │
|
||||
└──────┬───────┘
|
||||
4. Store the Event, a pending Delivery for each
|
||||
active Target, and the totals, in one
|
||||
transaction
|
||||
5. Build a delivery.Task for each Delivery
|
||||
(target settings + event data, the body
|
||||
inline when it is under 16 KiB)
|
||||
6. Queue the tasks on the delivery channel
|
||||
│
|
||||
┌── bounded worker pool (N workers) ──┐
|
||||
▼ ▼ ▼
|
||||
@@ -2180,6 +2188,12 @@ External Service
|
||||
└────────────┘
|
||||
```
|
||||
|
||||
A **Resubmit** stores its copy of an event through the same `Engine.StoreEvent`,
|
||||
and a **Replay** adds one delivery of a stored event through
|
||||
`Engine.CreateReplay`, which writes the delivery and its target's totals in one
|
||||
transaction and queues it. Those two methods are the only code that creates a
|
||||
delivery; the handlers read and check the request, then make the one call.
|
||||
|
||||
### Bounded Worker Pool
|
||||
|
||||
The delivery engine uses a **fixed-size worker pool** (default: 10 workers) to
|
||||
@@ -2189,8 +2203,9 @@ preventing goroutine explosions regardless of queue depth.
|
||||
**Architecture:**
|
||||
|
||||
- **Channels as queues:** Two buffered channels serve as bounded queues: a
|
||||
delivery channel (new tasks from the webhook handler) and a retry channel
|
||||
(tasks from backoff timers). Both are buffered to 10,000.
|
||||
delivery channel (the tasks of newly stored deliveries and of the `pending`
|
||||
deliveries the recovery routine hands back) and a retry channel (tasks from
|
||||
backoff timers). Both are buffered to 10,000.
|
||||
- **Fan-out via channel, not goroutines:** When an event arrives with multiple
|
||||
targets, each `delivery.Task` is sent to the delivery channel. Workers pick
|
||||
them up and process them — no goroutine-per-target.
|
||||
@@ -3054,6 +3069,7 @@ webhooker/
|
||||
│ │ └── logfield.go # Encoded-byte budget for client-supplied log values
|
||||
│ ├── delivery/
|
||||
│ │ ├── engine.go # Event-driven delivery engine (channel + timer based)
|
||||
│ │ ├── store.go # Storing an event with its deliveries, and a replay, then queueing them
|
||||
│ │ ├── circuit_breaker.go # Per-target circuit breaker for http/slack targets with retries
|
||||
│ │ ├── target.go # Target interface, Task, Scheduler
|
||||
│ │ ├── target_http.go # HTTP target (retries, circuit breaker)
|
||||
@@ -3150,23 +3166,26 @@ Components are wired via Uber fx in this order:
|
||||
10. `metrics.NewRegistry` — The registry `/metrics` serves
|
||||
11. `metrics.New` — The delivery collectors, registered on that registry
|
||||
12. `middleware.New` — HTTP middleware
|
||||
13. `delivery.New` — Event-driven delivery engine
|
||||
13. `delivery.New` — Event-driven delivery engine, which also stores each event
|
||||
with its deliveries and creates replay deliveries
|
||||
14. `archive.NewWriters` — The database targets' archive writers, and the lock
|
||||
every rename of their files runs under
|
||||
15. `archive.NewSweeper` — Periodic pruning of idle archives
|
||||
16. `delivery.Engine` → `delivery.Notifier` — interface bridge
|
||||
17. `archive.Writers` → `archive.Archives` — interface bridge so deleting or
|
||||
16. `archive.Writers` → `archive.Archives` — interface bridge so deleting or
|
||||
renaming a webhook or target reaches its archive files
|
||||
17. `delivery.Engine` → `delivery.CircuitBreakers` — interface bridge so the
|
||||
pages can show a target whose deliveries are paused
|
||||
18. `server.New` — HTTP server and router
|
||||
|
||||
The server starts via
|
||||
`fx.Invoke(func(*server.Server, *delivery.Engine, *database.RetentionReaper, *archive.Sweeper) {})`,
|
||||
which triggers the fx lifecycle hooks in dependency order. The
|
||||
`delivery.Notifier` interface allows the webhook handler to send self-contained
|
||||
`delivery.Task` slices to the engine without a direct package dependency. Each
|
||||
task carries all target settings and event data inline (for bodies under 16 KiB,
|
||||
`delivery.MaxInlineBodySize`), so the engine can deliver without reading from
|
||||
any database — it only writes to record results.
|
||||
which triggers the fx lifecycle hooks in dependency order. The handlers take
|
||||
`*delivery.Engine` itself: the receiver and resubmit handlers store an event
|
||||
through `StoreEvent`, and the replay handler adds a delivery through
|
||||
`CreateReplay`. Each task the engine queues carries the target's settings and
|
||||
the event's data, and the body too when it is under 16 KiB
|
||||
(`delivery.MaxInlineBodySize`); a larger body is read from the event row when
|
||||
the attempt runs.
|
||||
|
||||
### Middleware Stack
|
||||
|
||||
@@ -3466,8 +3485,8 @@ fixed independently of the compiler's:
|
||||
`golangci-lint config verify` and `golangci-lint run`, both with
|
||||
`--network=none`, and depends on `js-lint` (it copies a file from it).
|
||||
`make lint` builds this stage alone.
|
||||
5. **Test phase** (`test`, `golang:1.26.1-bookworm`, whose C compiler `-race`
|
||||
needs) — extracts Alpine.js, then runs `go test -race -cover` at most four
|
||||
5. **Test phase** (`test`, `golang:1.26.1-bookworm`, the builder stage's image)
|
||||
— extracts Alpine.js, then runs `go test -cover` with cgo off, at most four
|
||||
packages and eight tests at a time, with a 90-second timeout per package. On
|
||||
a failure it runs only the failed tests again with `-v`, since verbose output
|
||||
from the whole suite would pass the 2 MiB at which the Docker build cuts off
|
||||
|
||||
@@ -1,72 +1,77 @@
|
||||
# Workflow
|
||||
|
||||
One issue per unit of work, one branch and one PR per issue:
|
||||
One issue per unit of work, and one branch and one PR per issue:
|
||||
|
||||
- ensure a tracked issue exists with a definition of done
|
||||
- branch from `next` (never from `main`)
|
||||
- do the work; open a PR based on `next` (never on `main`)
|
||||
- pass an independent review, then the manager squash-merges into `next`
|
||||
- push; nothing stays local-only
|
||||
- make sure a tracked issue exists with a definition of done
|
||||
- branch from `next`, never from `main`
|
||||
- do the work and push the branch
|
||||
- open a PR with base `next`, never `main`
|
||||
- pass an independent review
|
||||
- the repository manager squash-merges it into `next`
|
||||
|
||||
`next` is the branch for the next milestone and must stay green and mergeable to
|
||||
`main` without notice. One `next` -> `main` PR accumulates the milestone;
|
||||
releases are cut from `main` separately.
|
||||
The repository manager is the role that merges each issue's PR into `next`.
|
||||
Nothing stays local-only: every commit is pushed.
|
||||
|
||||
The tracker (https://git.eeqj.de/sneak/webhooker/issues) is the record of what
|
||||
is done and what is under way. This file keeps no list of either.
|
||||
`next` is the branch for the next milestone. It must stay green and mergeable
|
||||
into `main` at all times. A milestone's work reaches `main` through one PR from
|
||||
`next`, which the owner, @sneak, merges. Cutting a release from `main` is a
|
||||
separate step after that.
|
||||
|
||||
The tracker (https://git.eeqj.de/sneak/webhooker/issues) records what is done
|
||||
and what is under way. This file keeps no list of either.
|
||||
|
||||
# Status
|
||||
|
||||
What is still open for 1.0.0 is in its milestone
|
||||
(https://git.eeqj.de/sneak/webhooker/milestone/9). `v1.0.0` is not tagged, and
|
||||
only @sneak tags it.
|
||||
(https://git.eeqj.de/sneak/webhooker/milestone/9). `v1.0.0` is not tagged. Only
|
||||
@sneak tags it.
|
||||
|
||||
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.
|
||||
Delivery is at least once by design. When webhooker cannot record the outcome of
|
||||
a delivery, it sends that delivery again. The target can then receive the same
|
||||
event twice. The alternative is a lost delivery that nothing reports. The README
|
||||
says so. Such a duplicate is not a defect to re-file.
|
||||
|
||||
# Next Step
|
||||
|
||||
@sneak runs webhooker in limited production, then tags `v1.0.0`. Merging `next`
|
||||
into `main` is a separate act from tagging and waits on neither of those: `next`
|
||||
is kept mergeable at all times, which is the point of the branch.
|
||||
into `main` is a separate act. It waits on neither the production run nor the
|
||||
tag.
|
||||
|
||||
# Future Steps
|
||||
|
||||
- Per-webhook rate limits, set in the web UI and enforced in the receiver
|
||||
handler, on top of the receiver's existing limits, which `RECEIVER_RATE_LIMIT`
|
||||
sets for the whole server (see Rate Limiting in the README)
|
||||
- API key authentication for programmatic access (the `APIKey` model exists, but
|
||||
nothing creates a key and no middleware checks one)
|
||||
- REST API v1
|
||||
- CRUD for webhooks, entrypoints, targets
|
||||
- Per-webhook rate limits, set in the web UI and enforced at the entrypoint.
|
||||
Today `RECEIVER_RATE_LIMIT` is the only rate-limit setting. It is how many
|
||||
requests per minute one client address may send to one entrypoint. The same
|
||||
address is also capped at ten times that across all entrypoints together.
|
||||
- API key authentication for programmatic access. The `APIKey` model and its
|
||||
`api_keys` table exist. Nothing creates a key and nothing checks one.
|
||||
- REST API v1. `/api/v1` is mounted with no routes today.
|
||||
- CRUD for webhooks, entrypoints and targets
|
||||
- event viewing and filtering endpoints
|
||||
- event redelivery endpoint
|
||||
- OpenAPI specification
|
||||
- Analytics in the web UI: response times summarised per webhook or per target,
|
||||
and figures for all webhooks together. The webhook page already shows event,
|
||||
delivery and failure counts and failure percentages, the target list counts
|
||||
each target's deliveries, the event log and each event's page show every
|
||||
attempt's duration, and `/metrics` carries server-wide counts and delivery
|
||||
durations
|
||||
- A remember-me option at login. Every session already survives a browser
|
||||
restart: it lasts up to 7 days, and ends sooner after `SESSION_IDLE_TIMEOUT`
|
||||
(default 24 hours) without activity
|
||||
- Password reset flow for a forgotten password over the web. The authenticated
|
||||
password _change_ flow already landed, and a lost password is recoverable from
|
||||
the console with `webhooker resetpw`
|
||||
(https://git.eeqj.de/sneak/webhooker/issues/208)
|
||||
and figures for all webhooks together. Today the webhook page shows event,
|
||||
delivery and failure counts and failure percentages. The target list counts
|
||||
each target's deliveries. The event log shows every attempt's duration.
|
||||
`/metrics` carries server-wide counts and delivery durations.
|
||||
- A remember-me option at login. Today every session survives a browser restart.
|
||||
It lasts up to 7 days. It ends sooner after `SESSION_IDLE_TIMEOUT` (default 24
|
||||
hours) without use.
|
||||
- Password reset over the web for a forgotten password. Today a signed-in user
|
||||
can change their password. A lost password is reset from the console with
|
||||
`webhooker resetpw` (https://git.eeqj.de/sneak/webhooker/issues/208).
|
||||
- Later, nice to have
|
||||
- email delivery target type
|
||||
- SNS and S3 delivery targets
|
||||
- data transformations (e.g. a configurable Slack message format; the Slack
|
||||
target sends one fixed format today)
|
||||
- JSONL file delivery with periodic S3 upload (the `database` target already
|
||||
archives events to SQLite files, rotated hourly, daily or monthly if set,
|
||||
and the webhook page offers them as one gzipped JSON download)
|
||||
- data transformations, such as a configurable Slack message format. The
|
||||
Slack target sends one fixed format today.
|
||||
- JSONL file delivery with periodic S3 upload. Today the `database` target
|
||||
archives events to SQLite files. A target can rotate those files hourly,
|
||||
daily or monthly. The webhook page offers them as one gzipped JSON
|
||||
download.
|
||||
- webhook event search, and filtering beyond the event log's failed and
|
||||
pending views
|
||||
- multi-user with role-based access control (each webhook already belongs to
|
||||
one account, but there is only the account made at first boot and no way
|
||||
to add another)
|
||||
- multi-user with role-based access control. Today each webhook belongs to
|
||||
one account. Only the account made at first boot exists. There is no way
|
||||
to add another.
|
||||
|
||||
@@ -211,9 +211,6 @@ func newApp() *fx.App {
|
||||
delivery.New,
|
||||
archive.NewWriters,
|
||||
archive.NewSweeper,
|
||||
// Wire *delivery.Engine as delivery.Notifier so the
|
||||
// webhook handler can notify the engine of new deliveries.
|
||||
func(e *delivery.Engine) delivery.Notifier { return e },
|
||||
// Wire *archive.Writers as archive.Archives so deleting
|
||||
// or renaming a webhook or target reaches its archive
|
||||
// files.
|
||||
|
||||
@@ -303,7 +303,7 @@ func (r *racingWrites) wait() (bool, error) {
|
||||
// contending for the writer's mutex when the eviction takes it.
|
||||
// The write that wins the mutex after the eviction must abandon
|
||||
// its work rather than reopen the archive, leaving the writer
|
||||
// permanently handle-free. Run under -race.
|
||||
// permanently handle-free.
|
||||
func TestEvictWebhook_RacingWriteDoesNotReopenHandle(
|
||||
t *testing.T,
|
||||
) {
|
||||
|
||||
@@ -959,9 +959,11 @@ func TestArchiveSweep_SkipsDeletedWebhookTargets(t *testing.T) {
|
||||
)
|
||||
}
|
||||
|
||||
// TestArchiveSweep_ConcurrentWrites proves the sweep serialises
|
||||
// against writes through the target's writer mutex. Run under
|
||||
// -race, an unsynchronised sweep would be caught here.
|
||||
// TestArchiveSweep_ConcurrentWrites runs the sweep while writes to
|
||||
// the same target are under way. The sweep serialises against them
|
||||
// through the target's writer mutex; the suite runs without Go's race
|
||||
// detector, so a sweep that did not would fail this test only by
|
||||
// panicking, hanging or leaving no archive file.
|
||||
func TestArchiveSweep_ConcurrentWrites(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -67,9 +67,10 @@ func DefaultPasswordConfig() *PasswordConfig {
|
||||
// testArgon2Memory is the Argon2id memory cost, in KiB, that a test
|
||||
// binary hashes with: 1 MB instead of the shipped 64 MB. Every test
|
||||
// that starts a database hashes the bootstrap admin password, dozens
|
||||
// of them run in parallel, and under the race detector each 64 MB hash
|
||||
// holds about 150 MB. VerifyPassword reads the cost from the hash it
|
||||
// checks, so verification follows.
|
||||
// of them run in parallel, and each 64 MB hash holds at least 64 MB
|
||||
// while it runs (about 150 MB when the suite ran under the race
|
||||
// detector). VerifyPassword reads the cost from the hash it checks, so
|
||||
// verification follows.
|
||||
const testArgon2Memory = 1024
|
||||
|
||||
// hashAtShippedCostInTest makes a test binary hash at the shipped
|
||||
|
||||
+60
-60
@@ -1,5 +1,5 @@
|
||||
// Package delivery manages asynchronous event delivery
|
||||
// to configured targets.
|
||||
// Package delivery stores each event with its deliveries and delivers
|
||||
// them, asynchronously, to configured targets.
|
||||
package delivery
|
||||
|
||||
import (
|
||||
@@ -23,10 +23,11 @@ import (
|
||||
|
||||
const (
|
||||
// deliveryChannelSize is the buffer size for the delivery
|
||||
// channel. New Tasks from the webhook handler are sent
|
||||
// here. Workers drain this channel. Sized large enough
|
||||
// that the webhook handler should never block under
|
||||
// normal load.
|
||||
// channel. The Tasks of newly stored deliveries, and of the
|
||||
// pending deliveries the recovery routine hands back, are
|
||||
// sent here. Workers drain this channel. Sized large enough
|
||||
// that no task should be dropped under normal load: one that
|
||||
// does not fit is left for the sweep.
|
||||
deliveryChannelSize = 10000
|
||||
|
||||
// retryChannelSize is the buffer size for the retry
|
||||
@@ -127,17 +128,11 @@ type Task struct {
|
||||
AttemptNum int
|
||||
}
|
||||
|
||||
// Notifier is the interface for notifying the delivery
|
||||
// engine about new deliveries.
|
||||
type Notifier interface {
|
||||
Notify(tasks []Task)
|
||||
}
|
||||
|
||||
// CircuitBreakers is how the handlers read a target's circuit
|
||||
// breaker, so the webhook page and the event log can say that
|
||||
// deliveries to the target are paused and until when. It keeps the
|
||||
// handlers free of the engine's internals and is trivially faked in
|
||||
// tests.
|
||||
// deliveries to the target are paused and until when. The handlers
|
||||
// hold the engine itself as well; they read breakers through this
|
||||
// interface so that a test can make one read as open.
|
||||
type CircuitBreakers interface {
|
||||
StateAndCooldown(targetID string) (CircuitState, time.Duration)
|
||||
}
|
||||
@@ -157,12 +152,13 @@ type EngineParams struct {
|
||||
}
|
||||
|
||||
// Engine processes queued deliveries in the background
|
||||
// using a bounded worker pool architecture. It owns only
|
||||
// the cross-target machinery: the worker pool, the queue
|
||||
// and retry channels, the recovery routine, and the persistence
|
||||
// helpers and Scheduler that individual targets rely on.
|
||||
// Each target type owns its own delivery, including retries,
|
||||
// backoff, and circuit breaking.
|
||||
// using a bounded worker pool architecture. It also stores each
|
||||
// event with its deliveries and creates replay deliveries (see
|
||||
// store.go). Beyond that it owns only the cross-target machinery:
|
||||
// the worker pool, the delivery and retry channels, the recovery
|
||||
// routine, and the persistence helpers and Scheduler that
|
||||
// individual targets rely on. Each target type owns its own
|
||||
// delivery, including retries, backoff, and circuit breaking.
|
||||
type Engine struct {
|
||||
database *database.Database
|
||||
dbManager *database.WebhookDBManager
|
||||
@@ -232,38 +228,6 @@ func New(
|
||||
return e
|
||||
}
|
||||
|
||||
// Notify signals the delivery engine that new deliveries
|
||||
// are ready.
|
||||
func (e *Engine) Notify(tasks []Task) {
|
||||
for i := range tasks {
|
||||
// Owned before it is queued, and until the worker that runs
|
||||
// it returns. A task can sit in a 10000-deep channel for a
|
||||
// long time on a healthy system, and nothing may re-send it
|
||||
// while it waits. See inflight.go.
|
||||
if !e.inflight.retainIdle(tasks[i].DeliveryID) {
|
||||
e.log.Warn(
|
||||
"delivery already in flight, not queued again",
|
||||
"delivery_id", tasks[i].DeliveryID,
|
||||
"event_id", tasks[i].EventID,
|
||||
)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
select {
|
||||
case e.deliveryCh <- tasks[i]:
|
||||
default:
|
||||
e.inflight.release(tasks[i].DeliveryID)
|
||||
e.log.Warn(
|
||||
"delivery channel full, "+
|
||||
"task will be recovered by the sweep",
|
||||
"delivery_id", tasks[i].DeliveryID,
|
||||
"event_id", tasks[i].EventID,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// StateAndCooldown implements CircuitBreakers. It returns the state of
|
||||
// the target's circuit breaker and, while the breaker is open, what is
|
||||
// left of its cooldown; the cooldown is zero once that has passed and
|
||||
@@ -324,6 +288,38 @@ func (e *Engine) ScheduleRetry(
|
||||
})
|
||||
}
|
||||
|
||||
// notify queues the tasks of deliveries already stored. StoreEvent and
|
||||
// CreateReplay call it once their transaction has committed.
|
||||
func (e *Engine) notify(tasks []Task) {
|
||||
for i := range tasks {
|
||||
// Owned before it is queued, and until the worker that runs
|
||||
// it returns. A task can sit in a 10000-deep channel for a
|
||||
// long time on a healthy system, and nothing may re-send it
|
||||
// while it waits. See inflight.go.
|
||||
if !e.inflight.retainIdle(tasks[i].DeliveryID) {
|
||||
e.log.Warn(
|
||||
"delivery already in flight, not queued again",
|
||||
"delivery_id", tasks[i].DeliveryID,
|
||||
"event_id", tasks[i].EventID,
|
||||
)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
select {
|
||||
case e.deliveryCh <- tasks[i]:
|
||||
default:
|
||||
e.inflight.release(tasks[i].DeliveryID)
|
||||
e.log.Warn(
|
||||
"delivery channel full, "+
|
||||
"task will be recovered by the sweep",
|
||||
"delivery_id", tasks[i].DeliveryID,
|
||||
"event_id", tasks[i].EventID,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// userAgent is the User-Agent header of every http and slack
|
||||
// delivery request: the program name and the running build's
|
||||
// version.
|
||||
@@ -472,9 +468,9 @@ func (e *Engine) processNewTask(
|
||||
}
|
||||
|
||||
// The recovery routine's run at start can send and release this
|
||||
// delivery before the receiver's Notify queues it. Ownership cannot
|
||||
// refuse a delivery nobody holds, so the row decides whether it
|
||||
// still needs sending.
|
||||
// delivery before StoreEvent or CreateReplay queues it. Ownership
|
||||
// cannot refuse a delivery nobody holds, so the row decides whether
|
||||
// it still needs sending.
|
||||
row, err := e.loadDelivery(webhookDB, task.DeliveryID)
|
||||
if err != nil {
|
||||
e.log.Error(
|
||||
@@ -899,7 +895,7 @@ func (e *Engine) recoverSingleRetry(
|
||||
return
|
||||
}
|
||||
|
||||
e.ScheduleRetry(buildRecoveryTask(
|
||||
e.ScheduleRetry(buildTask(
|
||||
d, webhookID, &event, &target, attemptNum+1,
|
||||
), remaining)
|
||||
|
||||
@@ -921,7 +917,7 @@ func (e *Engine) recoverSingleRetry(
|
||||
// it stopped working on before then: for example one whose bookkeeping
|
||||
// write failed, whose attempt a stop cut off, that was still queued
|
||||
// when the engine last stopped, whose task was dropped because the
|
||||
// delivery channel was full (by Notify or by queueRecovered), or whose
|
||||
// delivery channel was full (by notify or by queueRecovered), or whose
|
||||
// worker could not open the webhook's database.
|
||||
//
|
||||
// At start every pending delivery is taken, however recent, so one a
|
||||
@@ -1822,7 +1818,11 @@ func (e *Engine) loadTarget(
|
||||
return target, nil
|
||||
}
|
||||
|
||||
func buildRecoveryTask(
|
||||
// buildTask builds the task that carries delivery d of event to target
|
||||
// through the engine's channels, as attempt attemptNum. The body is
|
||||
// carried only when it is under MaxInlineBodySize; a larger one is read
|
||||
// from the event row when the attempt runs (see hydrateEvent).
|
||||
func buildTask(
|
||||
d *database.Delivery,
|
||||
webhookID string,
|
||||
event *database.Event,
|
||||
@@ -1998,7 +1998,7 @@ func (e *Engine) sendRecoveredDeliveries(
|
||||
continue
|
||||
}
|
||||
|
||||
task := buildRecoveryTask(
|
||||
task := buildTask(
|
||||
&deliveries[i], webhookID, &event, &target,
|
||||
attempts[deliveries[i].ID]+1,
|
||||
)
|
||||
|
||||
@@ -492,7 +492,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
|
||||
)
|
||||
task.TargetType = database.TargetTypeLog
|
||||
|
||||
s.Engine.Notify([]delivery.Task{task})
|
||||
s.Engine.ExportNotify([]delivery.Task{task})
|
||||
|
||||
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
||||
|
||||
@@ -1452,7 +1452,7 @@ func TestDeliverHTTP_NoURL(t *testing.T) {
|
||||
iAssertResultFailed(t, db, del.ID)
|
||||
}
|
||||
|
||||
// --- Notify batching ---
|
||||
// --- notify batching ---
|
||||
|
||||
func TestNotify_MultipleTasks(t *testing.T) {
|
||||
t.Parallel()
|
||||
@@ -1467,7 +1467,7 @@ func TestNotify_MultipleTasks(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
e.Notify(tasks)
|
||||
e.ExportNotify(tasks)
|
||||
|
||||
for i := range 5 {
|
||||
select {
|
||||
|
||||
@@ -29,7 +29,7 @@ const (
|
||||
// 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
|
||||
// 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.
|
||||
@@ -182,7 +182,7 @@ func TestEngine_WorkersOutliveStartHookContext(t *testing.T) {
|
||||
// routine's run at start cannot enqueue it.
|
||||
deliveryID, task := seedLogTask(t, s)
|
||||
|
||||
s.Engine.Notify([]delivery.Task{task})
|
||||
s.Engine.ExportNotify([]delivery.Task{task})
|
||||
|
||||
iWaitForDelivered(t, s.WebhookDB, deliveryID)
|
||||
}
|
||||
@@ -202,7 +202,7 @@ func TestEngine_StopHookStopsWorkers(t *testing.T) {
|
||||
// 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})
|
||||
s.Engine.ExportNotify([]delivery.Task{firstTask})
|
||||
iWaitForDelivered(t, s.WebhookDB, firstID)
|
||||
|
||||
var stopErr error
|
||||
@@ -232,7 +232,7 @@ func TestEngine_StopHookStopsWorkers(t *testing.T) {
|
||||
// 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})
|
||||
s.Engine.ExportNotify([]delivery.Task{secondTask})
|
||||
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
|
||||
@@ -383,7 +383,7 @@ func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) {
|
||||
t, s, 1, database.DeliveryStatusPending, 1,
|
||||
)
|
||||
|
||||
s.Engine.Notify([]delivery.Task{task})
|
||||
s.Engine.ExportNotify([]delivery.Task{task})
|
||||
stopDuringAttempt(t, lc, arrived)
|
||||
|
||||
iAssertStatus(
|
||||
@@ -448,7 +448,7 @@ func deliverToArchive(t *testing.T, s iSetup) string {
|
||||
task.TargetID = tgt.ID
|
||||
task.TargetType = database.TargetTypeDatabase
|
||||
|
||||
s.Engine.Notify([]delivery.Task{task})
|
||||
s.Engine.ExportNotify([]delivery.Task{task})
|
||||
|
||||
iWaitForDelivered(t, s.WebhookDB, deliveryID)
|
||||
|
||||
|
||||
@@ -218,7 +218,7 @@ func TestNotify_NonBlocking(t *testing.T) {
|
||||
done := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
e.Notify([]delivery.Task{
|
||||
e.ExportNotify([]delivery.Task{
|
||||
{DeliveryID: "overflow-1"},
|
||||
{DeliveryID: "overflow-2"},
|
||||
})
|
||||
@@ -229,7 +229,7 @@ func TestNotify_NonBlocking(t *testing.T) {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal(
|
||||
"Notify blocked when delivery channel was full",
|
||||
"notify blocked when delivery channel was full",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -158,6 +158,11 @@ func (e *Engine) ExportUpdateDeliveryStatus(
|
||||
return e.updateDeliveryStatus(webhookDB, d, "", status)
|
||||
}
|
||||
|
||||
// ExportNotify exposes notify.
|
||||
func (e *Engine) ExportNotify(tasks []Task) {
|
||||
e.notify(tasks)
|
||||
}
|
||||
|
||||
// ExportProcessNewTask exposes processNewTask.
|
||||
func (e *Engine) ExportProcessNewTask(
|
||||
ctx context.Context, task *Task,
|
||||
|
||||
@@ -61,7 +61,7 @@ func (s *inflightSet) retain(deliveryID string) {
|
||||
|
||||
// retainIdle takes the first reference on a delivery, and reports
|
||||
// whether it got it. It fails when the engine already owns the
|
||||
// delivery, which is what makes two claimants — Notify and the
|
||||
// delivery, which is what makes two claimants — notify and the
|
||||
// recovery routine, say — mutually exclusive rather than merely
|
||||
// atomic.
|
||||
func (s *inflightSet) retainIdle(deliveryID string) bool {
|
||||
|
||||
@@ -84,9 +84,9 @@ func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) {
|
||||
)
|
||||
rAgePending(t, s.WebhookDB, d.ID)
|
||||
|
||||
// Queued exactly as the receiver queues it, and never dequeued:
|
||||
// Queued exactly as StoreEvent queues it, and never dequeued:
|
||||
// no workers are running in this engine.
|
||||
s.Engine.Notify([]delivery.Task{{
|
||||
s.Engine.ExportNotify([]delivery.Task{{
|
||||
DeliveryID: d.ID,
|
||||
EventID: event.ID,
|
||||
WebhookID: s.WebhookID,
|
||||
@@ -109,7 +109,7 @@ func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) {
|
||||
|
||||
// TestRecoveryAndSweepDoNotDoubleDispatch drives the recovery routine's
|
||||
// run at start and a tick of the sweep at once against one aged pending
|
||||
// row. The engine runs them one after the other, but Notify can race
|
||||
// row. The engine runs them one after the other, but notify can race
|
||||
// either, and ownership is what keeps any two claimants from both
|
||||
// sending. Before ownership the two runs both dispatched the row.
|
||||
func TestRecoveryAndSweepDoNotDoubleDispatch(t *testing.T) {
|
||||
@@ -224,7 +224,7 @@ func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
|
||||
|
||||
body := `{"released":true}`
|
||||
|
||||
s.Engine.Notify([]delivery.Task{{
|
||||
s.Engine.ExportNotify([]delivery.Task{{
|
||||
DeliveryID: d.ID,
|
||||
EventID: event.ID,
|
||||
WebhookID: s.WebhookID,
|
||||
@@ -248,10 +248,10 @@ func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
|
||||
}
|
||||
|
||||
// TestNotifyAfterRecoveryDoesNotSendAgain is the startup race of
|
||||
// https://git.eeqj.de/sneak/webhooker/issues/299. The receiver has
|
||||
// https://git.eeqj.de/sneak/webhooker/issues/299. StoreEvent has
|
||||
// written a delivery, the recovery routine's run at start finds it
|
||||
// pending, sends it and releases it, and only then does the receiver's
|
||||
// Notify for it arrive. Nothing owns the delivery by then, so Notify
|
||||
// pending, sends it and releases it, and only then does StoreEvent's
|
||||
// notify for it arrive. Nothing owns the delivery by then, so notify
|
||||
// takes it.
|
||||
func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
|
||||
t.Parallel()
|
||||
@@ -289,7 +289,7 @@ func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
|
||||
|
||||
body := event.Body
|
||||
|
||||
s.Engine.Notify([]delivery.Task{{
|
||||
s.Engine.ExportNotify([]delivery.Task{{
|
||||
DeliveryID: d.ID,
|
||||
EventID: event.ID,
|
||||
WebhookID: s.WebhookID,
|
||||
@@ -300,7 +300,7 @@ func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
|
||||
EntrypointID: event.EntrypointID,
|
||||
}})
|
||||
|
||||
// Notify took the delivery, and a worker releases it once it has
|
||||
// notify took the delivery, and a worker releases it once it has
|
||||
// run the task.
|
||||
require.Eventually(
|
||||
t,
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
package delivery
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
)
|
||||
|
||||
// StoreEvent stores event with one pending delivery for each of its
|
||||
// webhook's active targets, and adds them to the totals, in one
|
||||
// transaction, then queues the deliveries. It returns how many
|
||||
// deliveries it created.
|
||||
//
|
||||
// The receiver stores every event it accepts through here, and a
|
||||
// resubmit stores its copy of a stored event, so every event is
|
||||
// stored, retried, SSRF-guarded and circuit-broken the same way however
|
||||
// it entered the system. CreateReplay is the only other place a
|
||||
// delivery is created.
|
||||
//
|
||||
// The targets are the webhook's active ones as they stand now: a
|
||||
// target created after a resubmitted event first arrived is included,
|
||||
// and an inactive one is skipped. An event with no active target is
|
||||
// stored with no deliveries.
|
||||
func (e *Engine) StoreEvent(event *database.Event) (int, error) {
|
||||
var targets []database.Target
|
||||
|
||||
err := e.database.DB().Where(
|
||||
"webhook_id = ? AND active = ?", event.WebhookID, true,
|
||||
).Find(&targets).Error
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("loading active targets: %w", err)
|
||||
}
|
||||
|
||||
webhookDB, err := e.dbManager.GetDB(event.WebhookID)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("getting webhook database: %w", err)
|
||||
}
|
||||
|
||||
event.BodyBytes = int64(len(event.Body))
|
||||
tasks := make([]Task, 0, len(targets))
|
||||
|
||||
err = webhookDB.Transaction(func(tx *gorm.DB) error {
|
||||
err := tx.Create(event).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("creating event: %w", err)
|
||||
}
|
||||
|
||||
for i := range targets {
|
||||
dlv, err := createDelivery(tx, event.ID, targets[i].ID, false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
tasks = append(tasks, buildTask(
|
||||
dlv, event.WebhookID, event, &targets[i], 1,
|
||||
))
|
||||
}
|
||||
|
||||
err = database.AddEventTotals(tx, database.EventTotals{
|
||||
Events: 1, LastEventAt: &event.CreatedAt,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// A resubmitted copy did not arrive on its entrypoint's URL, so
|
||||
// it leaves the entrypoint's last event as it is.
|
||||
if event.ResubmittedFromID != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
return database.AddEntrypointTotals(tx, database.EntrypointTotals{
|
||||
EntrypointID: event.EntrypointID, LastEventAt: event.CreatedAt,
|
||||
})
|
||||
})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
// Counted here, after the commit: an event exists once it is
|
||||
// durably stored, which is what the delivery counters are compared
|
||||
// against on a dashboard. A resubmitted event counts too, because
|
||||
// it produces deliveries that the delivery side counts; the
|
||||
// resubmit counter is what separates the two.
|
||||
e.mtr.EventReceived()
|
||||
|
||||
e.notify(tasks)
|
||||
|
||||
return len(tasks), nil
|
||||
}
|
||||
|
||||
// CreateReplay creates one more pending delivery of the stored event
|
||||
// eventID to target, marked as a replay, adds it to the target's totals
|
||||
// in the same transaction, and queues it. It returns the new
|
||||
// delivery's id.
|
||||
//
|
||||
// The caller passes the target as it stands now and has already
|
||||
// decided the replay may go ahead. The delivery being replayed is not
|
||||
// touched.
|
||||
func (e *Engine) CreateReplay(
|
||||
webhookID, eventID string, target *database.Target,
|
||||
) (string, error) {
|
||||
webhookDB, err := e.dbManager.GetDB(webhookID)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("getting webhook database: %w", err)
|
||||
}
|
||||
|
||||
event, err := e.loadEvent(webhookDB, eventID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
var dlv *database.Delivery
|
||||
|
||||
err = webhookDB.Transaction(func(tx *gorm.DB) error {
|
||||
var err error
|
||||
|
||||
dlv, err = createDelivery(tx, event.ID, target.ID, true)
|
||||
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
e.notify([]Task{buildTask(dlv, webhookID, &event, target, 1)})
|
||||
|
||||
return dlv.ID, nil
|
||||
}
|
||||
|
||||
// createDelivery writes a pending delivery of the event to the target
|
||||
// on the caller's transaction, and adds it to the target's totals.
|
||||
//
|
||||
// Neither Event nor Target is set on the row, and the per-webhook
|
||||
// database omits associations on every write in any case (see
|
||||
// internal/database/event_db_isolation.go), so the target's settings
|
||||
// are never copied into the event database.
|
||||
func createDelivery(
|
||||
tx *gorm.DB, eventID, targetID string, replay bool,
|
||||
) (*database.Delivery, error) {
|
||||
dlv := &database.Delivery{
|
||||
EventID: eventID,
|
||||
TargetID: targetID,
|
||||
Status: database.DeliveryStatusPending,
|
||||
Replay: replay,
|
||||
}
|
||||
|
||||
err := tx.Create(dlv).Error
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"creating delivery for target %s: %w", targetID, err,
|
||||
)
|
||||
}
|
||||
|
||||
err = database.AddTargetTotals(tx, database.TargetTotals{
|
||||
TargetID: targetID, Deliveries: 1,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return dlv, nil
|
||||
}
|
||||
@@ -0,0 +1,386 @@
|
||||
package delivery_test
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
"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"
|
||||
)
|
||||
|
||||
// storeTestTarget creates an active http target of the setup's webhook
|
||||
// with every one of a target's settings set, so a task that drops one
|
||||
// of them differs from the target.
|
||||
func storeTestTarget(
|
||||
t *testing.T, s iSetup, name string,
|
||||
) database.Target {
|
||||
t.Helper()
|
||||
|
||||
target := database.Target{
|
||||
WebhookID: s.WebhookID,
|
||||
Name: name,
|
||||
Type: database.TargetTypeHTTP,
|
||||
Active: true,
|
||||
URL: "http://93.184.216.34/" + name,
|
||||
Headers: map[string]string{"X-Api-Key": "k"},
|
||||
TimeoutSeconds: 7,
|
||||
ForwardQuery: true,
|
||||
ArchiveExpiry: time.Hour,
|
||||
ArchiveRotation: rotationDaily,
|
||||
MaxAttempts: 3,
|
||||
}
|
||||
|
||||
require.NoError(t, s.MainDB.Create(&target).Error)
|
||||
|
||||
return target
|
||||
}
|
||||
|
||||
// newStoreTestEvent returns an event of the setup's webhook as the
|
||||
// receiver builds one, not yet stored.
|
||||
func newStoreTestEvent(s iSetup, body string) *database.Event {
|
||||
return &database.Event{
|
||||
WebhookID: s.WebhookID,
|
||||
EntrypointID: uuid.New().String(),
|
||||
Method: http.MethodPost,
|
||||
RawQuery: "a=1&b=2",
|
||||
Headers: `{"X-Test":["yes"]}`,
|
||||
ContentType: testContentType,
|
||||
Body: body,
|
||||
}
|
||||
}
|
||||
|
||||
// queuedTasks takes every task the engine has queued so far. The
|
||||
// engine in these tests is never started, so nothing else takes them.
|
||||
func queuedTasks(e *delivery.Engine) []delivery.Task {
|
||||
var tasks []delivery.Task
|
||||
|
||||
for {
|
||||
select {
|
||||
case task := <-e.ExportDeliveryCh():
|
||||
tasks = append(tasks, task)
|
||||
default:
|
||||
return tasks
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// assertTaskFor proves task carries delivery deliveryID of event to
|
||||
// target: every one of the target's settings and every event field.
|
||||
// They are copied into the task one field at a time, and a field left
|
||||
// out is sent empty without any error.
|
||||
func assertTaskFor(
|
||||
t *testing.T,
|
||||
task delivery.Task,
|
||||
deliveryID string,
|
||||
event *database.Event,
|
||||
target *database.Target,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
assert.Equal(t, deliveryID, task.DeliveryID)
|
||||
assert.Equal(t, event.ID, task.EventID)
|
||||
assert.Equal(t, event.WebhookID, task.WebhookID)
|
||||
assert.Equal(t, event.EntrypointID, task.EntrypointID)
|
||||
|
||||
assert.Equal(t, target.ID, task.TargetID)
|
||||
assert.Equal(t, target.Name, task.TargetName)
|
||||
assert.Equal(t, target.Type, task.TargetType)
|
||||
assert.Equal(t, target.URL, task.TargetURL)
|
||||
assert.Equal(t, target.Headers, task.TargetHeaders)
|
||||
assert.Equal(t, target.TimeoutSeconds, task.TimeoutSeconds)
|
||||
assert.Equal(t, target.ForwardQuery, task.ForwardQuery)
|
||||
assert.Equal(t, target.ArchiveExpiry, task.ArchiveExpiry)
|
||||
assert.Equal(t, target.ArchiveRotation, task.ArchiveRotation)
|
||||
assert.Equal(t, target.MaxAttempts, task.MaxAttempts)
|
||||
|
||||
assert.Equal(t, event.Method, task.Method)
|
||||
assert.Equal(t, event.RawQuery, task.RawQuery)
|
||||
assert.Equal(t, event.Headers, task.Headers)
|
||||
assert.Equal(t, event.ContentType, task.ContentType)
|
||||
require.NotNil(t, task.Body)
|
||||
assert.Equal(t, event.Body, *task.Body)
|
||||
|
||||
assert.Equal(t, 1, task.AttemptNum)
|
||||
}
|
||||
|
||||
// storedDeliveries reads every delivery of an event.
|
||||
func storedDeliveries(
|
||||
t *testing.T, db *gorm.DB, eventID string,
|
||||
) []database.Delivery {
|
||||
t.Helper()
|
||||
|
||||
var deliveries []database.Delivery
|
||||
|
||||
require.NoError(t, db.Where("event_id = ?", eventID).
|
||||
Find(&deliveries).Error)
|
||||
|
||||
return deliveries
|
||||
}
|
||||
|
||||
// storedEventTotals reads the webhook's event totals.
|
||||
func storedEventTotals(t *testing.T, db *gorm.DB) database.EventTotals {
|
||||
t.Helper()
|
||||
|
||||
var totals database.EventTotals
|
||||
|
||||
require.NoError(t, db.First(&totals).Error)
|
||||
|
||||
return totals
|
||||
}
|
||||
|
||||
// countRows counts the rows of model in db.
|
||||
func countRows(t *testing.T, db *gorm.DB, model any) int64 {
|
||||
t.Helper()
|
||||
|
||||
var n int64
|
||||
|
||||
require.NoError(t, db.Model(model).Count(&n).Error)
|
||||
|
||||
return n
|
||||
}
|
||||
|
||||
// TestStoreEvent_StoresDeliveriesToActiveTargets proves StoreEvent
|
||||
// stores the event with one pending delivery to each active target and
|
||||
// none to an inactive one, adds them to the totals, and queues one task
|
||||
// per delivery.
|
||||
func TestStoreEvent_StoresDeliveriesToActiveTargets(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
targets := []database.Target{
|
||||
storeTestTarget(t, s, "first"),
|
||||
storeTestTarget(t, s, "second"),
|
||||
}
|
||||
off := storeTestTarget(t, s, "off")
|
||||
|
||||
require.NoError(t, s.MainDB.Model(&database.Target{}).
|
||||
Where("id = ?", off.ID).Update("active", false).Error)
|
||||
|
||||
event := newStoreTestEvent(s, `{"hello":"world"}`)
|
||||
|
||||
created, err := s.Engine.StoreEvent(event)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 2, created)
|
||||
|
||||
var stored database.Event
|
||||
|
||||
require.NoError(t, s.WebhookDB.First(&stored, "id = ?", event.ID).Error)
|
||||
assert.Equal(t, event.Body, stored.Body)
|
||||
assert.Equal(t, int64(len(event.Body)), stored.BodyBytes)
|
||||
|
||||
rows := storedDeliveries(t, s.WebhookDB, event.ID)
|
||||
queued := queuedTasks(s.Engine)
|
||||
|
||||
require.Len(t, rows, 2)
|
||||
require.Len(t, queued, 2)
|
||||
|
||||
deliveries := map[string]database.Delivery{}
|
||||
for _, dlv := range rows {
|
||||
deliveries[dlv.TargetID] = dlv
|
||||
}
|
||||
|
||||
tasks := map[string]delivery.Task{}
|
||||
for _, task := range queued {
|
||||
tasks[task.TargetID] = task
|
||||
}
|
||||
|
||||
for i := range targets {
|
||||
dlv, ok := deliveries[targets[i].ID]
|
||||
require.True(t, ok, "no delivery to %s", targets[i].Name)
|
||||
assert.Equal(t, database.DeliveryStatusPending, dlv.Status)
|
||||
assert.False(t, dlv.Replay)
|
||||
assert.Equal(t, int64(1),
|
||||
targetTotals(t, s.WebhookDB, targets[i].ID).Deliveries)
|
||||
|
||||
assertTaskFor(t, tasks[targets[i].ID], dlv.ID, event, &targets[i])
|
||||
}
|
||||
|
||||
totals := storedEventTotals(t, s.WebhookDB)
|
||||
assert.Equal(t, int64(1), totals.Events)
|
||||
require.NotNil(t, totals.LastEventAt)
|
||||
assert.True(t, totals.LastEventAt.Equal(stored.CreatedAt))
|
||||
|
||||
var entrypoint database.EntrypointTotals
|
||||
|
||||
require.NoError(t, s.WebhookDB.First(
|
||||
&entrypoint, "entrypoint_id = ?", event.EntrypointID,
|
||||
).Error)
|
||||
assert.True(t, entrypoint.LastEventAt.Equal(stored.CreatedAt))
|
||||
}
|
||||
|
||||
// TestStoreEvent_ResubmittedCopyLeavesTheEntrypoint proves a
|
||||
// resubmitted copy is counted among the webhook's events but leaves its
|
||||
// entrypoint's last event as it is: it did not arrive on that URL.
|
||||
func TestStoreEvent_ResubmittedCopyLeavesTheEntrypoint(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
storeTestTarget(t, s, "only")
|
||||
|
||||
sourceID := uuid.New().String()
|
||||
event := newStoreTestEvent(s, `{"copy":true}`)
|
||||
event.ResubmittedFromID = &sourceID
|
||||
|
||||
created, err := s.Engine.StoreEvent(event)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, created)
|
||||
assert.Len(t, queuedTasks(s.Engine), 1)
|
||||
|
||||
assert.Equal(t, int64(1), storedEventTotals(t, s.WebhookDB).Events)
|
||||
assert.Zero(t, countRows(t, s.WebhookDB, &database.EntrypointTotals{}))
|
||||
}
|
||||
|
||||
// TestStoreEvent_NoActiveTarget proves an event of a webhook with no
|
||||
// active target is stored with no delivery and nothing queued.
|
||||
func TestStoreEvent_NoActiveTarget(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
event := newStoreTestEvent(s, `{"to":"nobody"}`)
|
||||
|
||||
created, err := s.Engine.StoreEvent(event)
|
||||
require.NoError(t, err)
|
||||
assert.Zero(t, created)
|
||||
|
||||
assert.Equal(t, int64(1), countRows(t, s.WebhookDB, &database.Event{}))
|
||||
assert.Empty(t, storedDeliveries(t, s.WebhookDB, event.ID))
|
||||
assert.Empty(t, queuedTasks(s.Engine))
|
||||
}
|
||||
|
||||
// TestStoreEvent_LargeBodyIsNotCarried proves a body at the inline
|
||||
// limit is stored whole but left off the task, for the attempt to read
|
||||
// from the event row.
|
||||
func TestStoreEvent_LargeBodyIsNotCarried(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
storeTestTarget(t, s, "only")
|
||||
|
||||
event := newStoreTestEvent(
|
||||
s, strings.Repeat("x", delivery.MaxInlineBodySize),
|
||||
)
|
||||
|
||||
_, err := s.Engine.StoreEvent(event)
|
||||
require.NoError(t, err)
|
||||
|
||||
tasks := queuedTasks(s.Engine)
|
||||
require.Len(t, tasks, 1)
|
||||
assert.Nil(t, tasks[0].Body)
|
||||
|
||||
var stored database.Event
|
||||
|
||||
require.NoError(t, s.WebhookDB.First(&stored, "id = ?", event.ID).Error)
|
||||
assert.Equal(t, event.Body, stored.Body)
|
||||
}
|
||||
|
||||
// TestStoreEvent_RollsBackWhenAWriteFails proves the event, its
|
||||
// deliveries and its totals are one transaction: when the entrypoint
|
||||
// totals write fails, the last one, after the event, its delivery, the
|
||||
// target totals and the event totals were written, none of them is left
|
||||
// and nothing is queued.
|
||||
func TestStoreEvent_RollsBackWhenAWriteFails(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
storeTestTarget(t, s, "only")
|
||||
|
||||
require.NoError(t,
|
||||
s.WebhookDB.Migrator().DropTable(&database.EntrypointTotals{}))
|
||||
|
||||
_, err := s.Engine.StoreEvent(newStoreTestEvent(s, `{"lost":true}`))
|
||||
require.Error(t, err)
|
||||
|
||||
assert.Zero(t, countRows(t, s.WebhookDB, &database.Event{}))
|
||||
assert.Zero(t, countRows(t, s.WebhookDB, &database.Delivery{}))
|
||||
assert.Zero(t, countRows(t, s.WebhookDB, &database.TargetTotals{}))
|
||||
|
||||
totals := storedEventTotals(t, s.WebhookDB)
|
||||
assert.Zero(t, totals.Events)
|
||||
assert.Nil(t, totals.LastEventAt)
|
||||
|
||||
assert.Empty(t, queuedTasks(s.Engine))
|
||||
assert.Zero(t, s.Engine.ExportInflightHeld())
|
||||
}
|
||||
|
||||
// TestCreateReplay_CreatesAndQueuesAReplay proves CreateReplay adds a
|
||||
// new pending delivery marked as a replay, counts it in the target's
|
||||
// totals, queues its task built from the target it was given and the
|
||||
// stored event, and leaves the delivery it repeats as it was. The
|
||||
// stored event has every event field set, so a task that drops one of
|
||||
// them differs from the event.
|
||||
func TestCreateReplay_CreatesAndQueuesAReplay(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
target := storeTestTarget(t, s, "replayed")
|
||||
event := newStoreTestEvent(s, `{"replay":"me"}`)
|
||||
require.NoError(t, s.WebhookDB.Create(event).Error)
|
||||
|
||||
original := iSeedDelivery(
|
||||
t, s.WebhookDB, event.ID, target.ID,
|
||||
database.DeliveryStatusFailed,
|
||||
)
|
||||
|
||||
deliveryID, err := s.Engine.CreateReplay(s.WebhookID, event.ID, &target)
|
||||
require.NoError(t, err)
|
||||
|
||||
replay := storedDelivery(t, s.WebhookDB, deliveryID)
|
||||
assert.Equal(t, event.ID, replay.EventID)
|
||||
assert.Equal(t, target.ID, replay.TargetID)
|
||||
assert.Equal(t, database.DeliveryStatusPending, replay.Status)
|
||||
assert.True(t, replay.Replay)
|
||||
|
||||
assert.Equal(t, database.DeliveryStatusFailed,
|
||||
storedDelivery(t, s.WebhookDB, original.ID).Status)
|
||||
assert.Equal(t, int64(1),
|
||||
targetTotals(t, s.WebhookDB, target.ID).Deliveries)
|
||||
|
||||
tasks := queuedTasks(s.Engine)
|
||||
require.Len(t, tasks, 1)
|
||||
assertTaskFor(t, tasks[0], deliveryID, event, &target)
|
||||
}
|
||||
|
||||
// TestCreateReplay_RollsBackWhenAWriteFails proves the replay delivery
|
||||
// and its target totals are one transaction: when the totals write
|
||||
// fails, after the delivery was written, no new delivery is left and
|
||||
// nothing is queued.
|
||||
func TestCreateReplay_RollsBackWhenAWriteFails(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
target := storeTestTarget(t, s, "replayed")
|
||||
event := iSeedEvent(t, s.WebhookDB, s.WebhookID, `{"replay":"me"}`)
|
||||
iSeedDelivery(
|
||||
t, s.WebhookDB, event.ID, target.ID,
|
||||
database.DeliveryStatusFailed,
|
||||
)
|
||||
|
||||
require.NoError(t,
|
||||
s.WebhookDB.Migrator().DropTable(&database.TargetTotals{}))
|
||||
|
||||
_, err := s.Engine.CreateReplay(s.WebhookID, event.ID, &target)
|
||||
require.Error(t, err)
|
||||
|
||||
assert.Len(t, storedDeliveries(t, s.WebhookDB, event.ID), 1)
|
||||
assert.Empty(t, queuedTasks(s.Engine))
|
||||
assert.Zero(t, s.Engine.ExportInflightHeld())
|
||||
}
|
||||
|
||||
// storedDelivery reads one delivery back.
|
||||
func storedDelivery(
|
||||
t *testing.T, db *gorm.DB, deliveryID string,
|
||||
) database.Delivery {
|
||||
t.Helper()
|
||||
|
||||
var dlv database.Delivery
|
||||
|
||||
require.NoError(t, db.First(&dlv, "id = ?", deliveryID).Error)
|
||||
|
||||
return dlv
|
||||
}
|
||||
@@ -5,9 +5,7 @@ import (
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/delivery"
|
||||
)
|
||||
|
||||
// The outcomes of a replay POST, as the notice codes its redirect
|
||||
@@ -46,12 +44,12 @@ const (
|
||||
// HandleDeliveryReplay re-sends a finished delivery's event to its
|
||||
// target.
|
||||
//
|
||||
// A replay never touches the delivery it repeats. It creates a NEW
|
||||
// pending delivery row for the same event and target and hands it to
|
||||
// the delivery engine through the same Notifier the receiver uses, so
|
||||
// the original's status, attempts and timestamps stand as the record
|
||||
// of what actually happened, and the replay is retried, SSRF-guarded
|
||||
// and circuit-broken exactly as a first attempt is.
|
||||
// A replay never touches the delivery it repeats. The delivery engine
|
||||
// creates a NEW pending delivery row for the same event and target and
|
||||
// queues it as it queues the receiver's, so the original's status,
|
||||
// attempts and timestamps stand as the record of what actually
|
||||
// happened, and the replay is retried, SSRF-guarded and circuit-broken
|
||||
// exactly as a first attempt is.
|
||||
//
|
||||
// What is re-sent is the stored EVENT body, never the response the
|
||||
// original delivery received.
|
||||
@@ -145,7 +143,8 @@ func (h *Handlers) loadReplaySource(
|
||||
return &original, true
|
||||
}
|
||||
|
||||
// queueReplay writes the new delivery and hands it to the engine.
|
||||
// queueReplay has the delivery engine create and queue the new
|
||||
// delivery, unless an earlier replay is still in flight.
|
||||
func (h *Handlers) queueReplay(
|
||||
w http.ResponseWriter,
|
||||
r *http.Request,
|
||||
@@ -171,18 +170,8 @@ func (h *Handlers) queueReplay(
|
||||
return
|
||||
}
|
||||
|
||||
var event database.Event
|
||||
|
||||
err = webhookDB.
|
||||
First(&event, "id = ?", original.EventID).Error
|
||||
if err != nil {
|
||||
h.serverError(w, r, "failed to load event for replay", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
task, err := createReplayDelivery(
|
||||
webhookDB, webhook.ID, &event, target,
|
||||
deliveryID, err := h.engine.CreateReplay(
|
||||
webhook.ID, original.EventID, target,
|
||||
)
|
||||
if err != nil {
|
||||
h.serverError(
|
||||
@@ -193,15 +182,14 @@ func (h *Handlers) queueReplay(
|
||||
}
|
||||
|
||||
h.mtr.DeliveryReplayed(target.Type)
|
||||
h.notifier.Notify([]delivery.Task{task})
|
||||
|
||||
h.log.Info(
|
||||
"delivery replay queued",
|
||||
"webhook_id", webhook.ID,
|
||||
"event_id", event.ID,
|
||||
"event_id", original.EventID,
|
||||
"target_id", target.ID,
|
||||
"replayed_delivery_id", original.ID,
|
||||
"delivery_id", task.DeliveryID,
|
||||
"delivery_id", deliveryID,
|
||||
)
|
||||
|
||||
redirectToEventLog(w, r, webhook, replayQueued)
|
||||
@@ -263,77 +251,6 @@ func countInFlightDeliveries(
|
||||
return count, err
|
||||
}
|
||||
|
||||
// createReplayDelivery writes the new pending delivery row, adds it to
|
||||
// its target's totals in the same transaction, and returns the task
|
||||
// that carries it to the delivery engine.
|
||||
//
|
||||
// The row is written with associations omitted, and neither Event nor
|
||||
// Target is populated on it: GORM's SaveBeforeAssociations would
|
||||
// otherwise 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 createReplayDelivery(
|
||||
webhookDB *gorm.DB,
|
||||
webhookID string,
|
||||
event *database.Event,
|
||||
target *database.Target,
|
||||
) (delivery.Task, error) {
|
||||
dlv := &database.Delivery{
|
||||
EventID: event.ID,
|
||||
TargetID: target.ID,
|
||||
Status: database.DeliveryStatusPending,
|
||||
Replay: true,
|
||||
}
|
||||
|
||||
err := webhookDB.Transaction(func(tx *gorm.DB) error {
|
||||
err := tx.Omit(clause.Associations).Create(dlv).Error
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return database.AddTargetTotals(tx, database.TargetTotals{
|
||||
TargetID: dlv.TargetID, Deliveries: 1,
|
||||
})
|
||||
})
|
||||
if err != nil {
|
||||
return delivery.Task{}, err
|
||||
}
|
||||
|
||||
return delivery.Task{
|
||||
DeliveryID: dlv.ID,
|
||||
EventID: event.ID,
|
||||
WebhookID: webhookID,
|
||||
EntrypointID: event.EntrypointID,
|
||||
TargetID: target.ID,
|
||||
TargetName: target.Name,
|
||||
TargetType: target.Type,
|
||||
TargetURL: target.URL,
|
||||
TargetHeaders: target.Headers,
|
||||
TimeoutSeconds: target.TimeoutSeconds,
|
||||
ForwardQuery: target.ForwardQuery,
|
||||
ArchiveExpiry: target.ArchiveExpiry,
|
||||
ArchiveRotation: target.ArchiveRotation,
|
||||
MaxAttempts: target.MaxAttempts,
|
||||
Method: event.Method,
|
||||
RawQuery: event.RawQuery,
|
||||
Headers: event.Headers,
|
||||
ContentType: event.ContentType,
|
||||
Body: replayBody(event.Body),
|
||||
AttemptNum: 1,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// replayBody returns the stored event body for a replay task to carry
|
||||
// inline, or nil when it is large enough that the engine should fetch
|
||||
// it from the per-webhook database instead.
|
||||
func replayBody(body string) *string {
|
||||
if len(body) >= delivery.MaxInlineBodySize {
|
||||
return nil
|
||||
}
|
||||
|
||||
return &body
|
||||
}
|
||||
|
||||
// redirectToEventLog redirects a replay or resubmit back to the event
|
||||
// log it was triggered from, carrying the outcome as its notice. A
|
||||
// Replay form carries the list it was pressed in as show, so a replay
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/delivery"
|
||||
"sneak.berlin/go/webhooker/internal/handlers"
|
||||
"sneak.berlin/go/webhooker/internal/session"
|
||||
)
|
||||
@@ -23,7 +22,7 @@ const paramDeliveryID = "deliveryID"
|
||||
|
||||
// replayTargetURL is a public destination, so a target configured with
|
||||
// it is one the SSRF guard would accept. Nothing in these tests
|
||||
// dispatches to it: the notifier is recorded, not run.
|
||||
// dispatches to it: the delivery engine is never started.
|
||||
const replayTargetURL = "http://93.184.216.34/hook"
|
||||
|
||||
// replayEventQuery is the query string a seeded event arrived with.
|
||||
@@ -163,14 +162,8 @@ func postReplay(
|
||||
|
||||
// TestHandleDeliveryReplay_AppendsDeliveryAndLeavesOriginal is the
|
||||
// core requirement: replaying a failed delivery succeeds, appends a
|
||||
// new delivery, and leaves the original row and its recorded attempt
|
||||
// exactly as they were.
|
||||
//
|
||||
// It also pins the two things a replay would be wrong to get from the
|
||||
// original: the task carries the target's CURRENT configuration, which
|
||||
// this test changes between the failure and the replay, and it carries
|
||||
// the stored EVENT body rather than anything the failed attempt
|
||||
// received back.
|
||||
// new delivery marked as a replay, and leaves the original row and its
|
||||
// recorded attempt exactly as they were.
|
||||
func TestHandleDeliveryReplay_AppendsDeliveryAndLeavesOriginal(
|
||||
t *testing.T,
|
||||
) {
|
||||
@@ -181,10 +174,9 @@ func TestHandleDeliveryReplay_AppendsDeliveryAndLeavesOriginal(
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -204,15 +196,6 @@ func TestHandleDeliveryReplay_AppendsDeliveryAndLeavesOriginal(
|
||||
|
||||
before := loadDelivery(t, webhookDB, original.ID)
|
||||
|
||||
// The operator fixes the destination, which is the whole reason
|
||||
// to replay. The replay must use this, not the URL the original
|
||||
// delivery ran against.
|
||||
const fixedURL = "http://93.184.216.34/fixed"
|
||||
|
||||
require.NoError(t, db.DB().Model(&database.Target{}).
|
||||
Where("id = ?", tgt.ID).
|
||||
Update("url", fixedURL).Error)
|
||||
|
||||
w := postReplay(t, h, sess, wh.ID, original.ID)
|
||||
|
||||
require.Equal(t, http.StatusSeeOther, w.Code)
|
||||
@@ -234,14 +217,9 @@ func TestHandleDeliveryReplay_AppendsDeliveryAndLeavesOriginal(
|
||||
assert.Equal(
|
||||
t, database.DeliveryStatusPending, replayed.Status,
|
||||
)
|
||||
assert.True(t, replayed.Replay)
|
||||
|
||||
assertDeliveryUntouched(t, webhookDB, before)
|
||||
|
||||
tasks := notif.Tasks()
|
||||
require.Len(t, tasks, 1)
|
||||
assertReplayTask(
|
||||
t, tasks[0], wh.ID, event, tgt, replayed.ID, fixedURL,
|
||||
)
|
||||
assertNoLeakedTarget(t, webhookDB)
|
||||
}
|
||||
|
||||
@@ -275,45 +253,6 @@ func assertDeliveryUntouched(
|
||||
)
|
||||
}
|
||||
|
||||
// assertReplayTask proves the task handed to the delivery engine is
|
||||
// the one the receiver would build for this event and this target, and
|
||||
// that it carries wantURL — the target's URL as it stands now rather
|
||||
// than as the original delivery ran against it.
|
||||
func assertReplayTask(
|
||||
t *testing.T,
|
||||
task delivery.Task,
|
||||
webhookID string,
|
||||
event *database.Event,
|
||||
target *database.Target,
|
||||
wantDeliveryID, wantURL string,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
assert.Equal(t, wantDeliveryID, task.DeliveryID)
|
||||
assert.Equal(t, event.ID, task.EventID)
|
||||
assert.Equal(t, webhookID, task.WebhookID)
|
||||
assert.Equal(t, event.EntrypointID, task.EntrypointID)
|
||||
assert.Equal(t, target.ID, task.TargetID)
|
||||
assert.Equal(t, target.Type, task.TargetType)
|
||||
assert.Equal(
|
||||
t, wantURL, task.TargetURL,
|
||||
"replay must use the target's current URL",
|
||||
)
|
||||
assert.Equal(t, event.Method, task.Method)
|
||||
assert.Equal(
|
||||
t, replayEventQuery, task.RawQuery,
|
||||
"replay re-sends the stored query string",
|
||||
)
|
||||
assert.Equal(t, event.Headers, task.Headers)
|
||||
assert.Equal(t, event.ContentType, task.ContentType)
|
||||
assert.Equal(t, 1, task.AttemptNum)
|
||||
require.NotNil(t, task.Body)
|
||||
assert.Equal(
|
||||
t, event.Body, *task.Body,
|
||||
"replay re-sends the stored event body",
|
||||
)
|
||||
}
|
||||
|
||||
// assertNoLeakedTarget proves the per-webhook database holds no
|
||||
// target: an association write would have upserted the whole target,
|
||||
// plaintext config and all, into the event database. See
|
||||
@@ -327,12 +266,10 @@ func assertNoLeakedTarget(t *testing.T, webhookDB *gorm.DB) {
|
||||
)
|
||||
}
|
||||
|
||||
// TestHandleDeliveryReplay_TaskCarriesEveryTargetSetting proves the
|
||||
// task a replay hands the delivery engine carries every one of the
|
||||
// target's settings.
|
||||
func TestHandleDeliveryReplay_TaskCarriesEveryTargetSetting(
|
||||
t *testing.T,
|
||||
) {
|
||||
// TestHandleDeliveryReplay_FailedStoreIsAnError proves a replay the
|
||||
// engine cannot store answers with the 500 error page and adds no
|
||||
// delivery.
|
||||
func TestHandleDeliveryReplay_FailedStoreIsAnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
@@ -340,10 +277,9 @@ func TestHandleDeliveryReplay_TaskCarriesEveryTargetSetting(
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -351,22 +287,28 @@ func TestHandleDeliveryReplay_TaskCarriesEveryTargetSetting(
|
||||
wh := seedWebhook(t, db)
|
||||
tgt := seedConfiguredTarget(
|
||||
t, db, wh.ID, database.TargetTypeHTTP,
|
||||
targetWithEverySetting(),
|
||||
database.Target{URL: replayTargetURL},
|
||||
)
|
||||
|
||||
_, original := seedFailedDelivery(t, dbMgr, wh.ID, tgt.ID)
|
||||
event, original := seedFailedDelivery(t, dbMgr, wh.ID, tgt.ID)
|
||||
|
||||
// Without its target totals table the webhook's database refuses
|
||||
// the totals write, which comes after the delivery row.
|
||||
webhookDB, err := dbMgr.GetDB(wh.ID)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t,
|
||||
webhookDB.Migrator().DropTable(&database.TargetTotals{}))
|
||||
|
||||
w := postReplay(t, h, sess, wh.ID, original.ID)
|
||||
require.Equal(t, http.StatusSeeOther, w.Code)
|
||||
|
||||
tasks := notif.Tasks()
|
||||
require.Len(t, tasks, 1)
|
||||
assertTaskCarriesTargetSettings(t, tasks[0], tgt)
|
||||
assert.Equal(t, http.StatusInternalServerError, w.Code)
|
||||
assert.Contains(t, w.Body.String(), "500 Internal Server Error")
|
||||
assert.Len(t, listDeliveries(t, webhookDB, event.ID), 1)
|
||||
}
|
||||
|
||||
// TestHandleDeliveryReplay_RefusesDeletedTarget proves the required
|
||||
// refusal: a target deleted since the delivery ran is reported as
|
||||
// deleted rather than erroring, and nothing is created or queued.
|
||||
// deleted rather than erroring, and nothing is created.
|
||||
func TestHandleDeliveryReplay_RefusesDeletedTarget(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -375,10 +317,9 @@ func TestHandleDeliveryReplay_RefusesDeletedTarget(t *testing.T) {
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -414,10 +355,6 @@ func TestHandleDeliveryReplay_RefusesDeletedTarget(t *testing.T) {
|
||||
t, listDeliveries(t, webhookDB, event.ID), 1,
|
||||
"a refused replay must create no delivery",
|
||||
)
|
||||
assert.Empty(
|
||||
t, notif.Tasks(),
|
||||
"a refused replay must queue nothing",
|
||||
)
|
||||
|
||||
// The refusal is specific, which is why the target is looked up
|
||||
// including soft-deleted rows: an id that never named a target
|
||||
@@ -450,10 +387,9 @@ func TestHandleDeliveryReplay_RefusesWhileEarlierReplayInFlight(
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -491,10 +427,6 @@ func TestHandleDeliveryReplay_RefusesWhileEarlierReplayInFlight(
|
||||
t, listDeliveries(t, webhookDB, event.ID), 2,
|
||||
"the refused second replay must add nothing",
|
||||
)
|
||||
assert.Len(
|
||||
t, notif.Tasks(), 1,
|
||||
"only the first replay reaches the delivery engine",
|
||||
)
|
||||
|
||||
// A delivery the engine has not finished is not replayable
|
||||
// either, which is the same rule seen from the other side.
|
||||
|
||||
@@ -171,37 +171,29 @@ func loadResubmitSource(
|
||||
return src, true, nil
|
||||
}
|
||||
|
||||
// queueResubmit stores the copy and fans it out to the webhook's
|
||||
// active targets.
|
||||
// queueResubmit has the delivery engine store the copy, with a delivery
|
||||
// for each of the webhook's active targets as they stand now, through
|
||||
// the same StoreEvent the receiver uses. The one difference is that the
|
||||
// copy leaves its entrypoint's last event as it is, since it did not
|
||||
// arrive on that URL.
|
||||
func (h *Handlers) queueResubmit(
|
||||
w http.ResponseWriter,
|
||||
r *http.Request,
|
||||
webhook database.Webhook,
|
||||
src resubmitSource,
|
||||
) {
|
||||
// The receiver's own query, run now: an active target created
|
||||
// after the original event arrived is included, and an
|
||||
// inactive one is skipped rather than refused.
|
||||
targets, err := h.loadActiveTargets(webhook.ID)
|
||||
if err != nil {
|
||||
h.serverError(w, r, "failed to query targets", err)
|
||||
|
||||
return
|
||||
event := &database.Event{
|
||||
WebhookID: webhook.ID,
|
||||
EntrypointID: src.EntrypointID,
|
||||
Method: src.Method,
|
||||
RawQuery: src.RawQuery,
|
||||
Headers: src.Headers,
|
||||
ContentType: src.ContentType,
|
||||
Body: string(src.Body),
|
||||
ResubmittedFromID: &src.ID,
|
||||
}
|
||||
|
||||
event, tasks, err := h.createAndFanOut(
|
||||
eventSource{
|
||||
WebhookID: webhook.ID,
|
||||
EntrypointID: src.EntrypointID,
|
||||
Method: src.Method,
|
||||
RawQuery: src.RawQuery,
|
||||
HeadersJSON: src.Headers,
|
||||
ContentType: src.ContentType,
|
||||
Body: src.Body,
|
||||
ResubmittedFromID: &src.ID,
|
||||
},
|
||||
targets,
|
||||
)
|
||||
deliveries, err := h.engine.StoreEvent(event)
|
||||
if err != nil {
|
||||
h.serverError(w, r, "failed to store resubmitted event", err)
|
||||
|
||||
@@ -215,11 +207,11 @@ func (h *Handlers) queueResubmit(
|
||||
"webhook_id", webhook.ID,
|
||||
"event_id", event.ID,
|
||||
"resubmitted_from_id", src.ID,
|
||||
"target_count", len(tasks),
|
||||
"target_count", deliveries,
|
||||
)
|
||||
|
||||
code := resubmitQueued
|
||||
if len(tasks) == 0 {
|
||||
if deliveries == 0 {
|
||||
code = resubmitNoTargets
|
||||
}
|
||||
|
||||
|
||||
@@ -19,7 +19,7 @@ import (
|
||||
|
||||
// resubmitTargetURL is a public destination, so a target configured
|
||||
// with it is one the SSRF guard would accept. Nothing in these tests
|
||||
// dispatches to it: the notifier is recorded, not run.
|
||||
// dispatches to it: the delivery engine is never started.
|
||||
const resubmitTargetURL = "http://93.184.216.34/hook"
|
||||
|
||||
// resubmitEventHeaders and resubmitEventQuery are the stored header
|
||||
@@ -134,10 +134,9 @@ func TestHandleEventResubmit_DeliversToTargetCreatedAfterTheEvent(
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -189,11 +188,6 @@ func TestHandleEventResubmit_DeliversToTargetCreatedAfterTheEvent(
|
||||
t, database.DeliveryStatusPending, deliveries[0].Status,
|
||||
)
|
||||
|
||||
tasks := notif.Tasks()
|
||||
require.Len(t, tasks, 1)
|
||||
assert.Equal(t, deliveries[0].ID, tasks[0].DeliveryID)
|
||||
assertResubmitTask(t, tasks[0], wh.ID, &fresh, tgt)
|
||||
|
||||
assertNoLeakedTarget(t, webhookDB)
|
||||
}
|
||||
|
||||
@@ -225,32 +219,6 @@ func assertEventCopy(
|
||||
)
|
||||
}
|
||||
|
||||
// assertResubmitTask proves the task handed to the delivery engine is
|
||||
// the one the receiver would build for the NEW event and this target.
|
||||
func assertResubmitTask(
|
||||
t *testing.T,
|
||||
task delivery.Task,
|
||||
webhookID string,
|
||||
fresh *database.Event,
|
||||
target *database.Target,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
assert.Equal(t, fresh.ID, task.EventID)
|
||||
assert.Equal(t, webhookID, task.WebhookID)
|
||||
assert.Equal(t, fresh.EntrypointID, task.EntrypointID)
|
||||
assert.Equal(t, target.ID, task.TargetID)
|
||||
assert.Equal(t, target.Type, task.TargetType)
|
||||
assert.Equal(t, fresh.Method, task.Method)
|
||||
assert.Equal(t, fresh.RawQuery, task.RawQuery)
|
||||
assert.Equal(t, fresh.Headers, task.Headers)
|
||||
assert.Equal(t, fresh.ContentType, task.ContentType)
|
||||
assert.Equal(t, 1, task.AttemptNum)
|
||||
|
||||
require.NotNil(t, task.Body)
|
||||
assert.Equal(t, fresh.Body, *task.Body)
|
||||
}
|
||||
|
||||
// TestHandleEventResubmit_IsRepeatable proves the requirement replay
|
||||
// deliberately does not meet: firing the same captured event at a
|
||||
// backend over and over must work, with no in-flight refusal, and each
|
||||
@@ -263,10 +231,9 @@ func TestHandleEventResubmit_IsRepeatable(t *testing.T) {
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -300,30 +267,18 @@ func TestHandleEventResubmit_IsRepeatable(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
|
||||
events := listEvents(t, webhookDB)
|
||||
require.Len(t, events, presses+1)
|
||||
require.Len(
|
||||
t, events, presses+1, "each resubmit creates its own event",
|
||||
)
|
||||
|
||||
tasks := notif.Tasks()
|
||||
require.Len(t, tasks, presses)
|
||||
assert.Empty(
|
||||
t, listDeliveries(t, webhookDB, original.ID),
|
||||
"each resubmit delivers its own new event",
|
||||
)
|
||||
|
||||
seen := make(map[string]struct{}, presses)
|
||||
|
||||
for _, task := range tasks {
|
||||
assert.Equal(t, tgt.ID, task.TargetID)
|
||||
assert.NotEqual(
|
||||
t, original.ID, task.EventID,
|
||||
"each resubmit delivers its own new event",
|
||||
)
|
||||
|
||||
_, dup := seen[task.EventID]
|
||||
assert.False(t, dup, "each resubmit creates its own event")
|
||||
|
||||
seen[task.EventID] = struct{}{}
|
||||
|
||||
require.Len(t, listDeliveries(t, webhookDB, task.EventID), 1)
|
||||
}
|
||||
|
||||
// Every copy names the same source event, so twenty presses stay
|
||||
// traceable to the one captured request.
|
||||
// Every copy has its own delivery, and names the same source
|
||||
// event, so twenty presses stay traceable to the one captured
|
||||
// request.
|
||||
for _, e := range events {
|
||||
if e.ID == original.ID {
|
||||
continue
|
||||
@@ -331,6 +286,10 @@ func TestHandleEventResubmit_IsRepeatable(t *testing.T) {
|
||||
|
||||
require.NotNil(t, e.ResubmittedFromID)
|
||||
assert.Equal(t, original.ID, *e.ResubmittedFromID)
|
||||
|
||||
deliveries := listDeliveries(t, webhookDB, e.ID)
|
||||
require.Len(t, deliveries, 1)
|
||||
assert.Equal(t, tgt.ID, deliveries[0].TargetID)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -346,10 +305,9 @@ func TestHandleEventResubmit_OversizeBodySurvivesIntact(t *testing.T) {
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -385,22 +343,7 @@ func TestHandleEventResubmit_OversizeBodySurvivesIntact(t *testing.T) {
|
||||
t, stored, fresh.Body,
|
||||
"the stored body must be copied byte for byte",
|
||||
)
|
||||
|
||||
tasks := notif.Tasks()
|
||||
require.Len(t, tasks, 1)
|
||||
assert.Nil(
|
||||
t, tasks[0].Body,
|
||||
"a body over the inline limit is fetched from the new "+
|
||||
"event row rather than carried on the task",
|
||||
)
|
||||
|
||||
// The engine's own read of the body, against the new event id:
|
||||
// what it would send is what was stored.
|
||||
var loaded database.Event
|
||||
|
||||
require.NoError(t, webhookDB.Select("body").
|
||||
First(&loaded, "id = ?", tasks[0].EventID).Error)
|
||||
assert.Equal(t, stored, loaded.Body)
|
||||
assert.Len(t, listDeliveries(t, webhookDB, fresh.ID), 1)
|
||||
}
|
||||
|
||||
// TestHandleEventResubmit_SkipsInactiveTarget proves a deactivated
|
||||
@@ -414,10 +357,9 @@ func TestHandleEventResubmit_SkipsInactiveTarget(t *testing.T) {
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -448,21 +390,19 @@ func TestHandleEventResubmit_SkipsInactiveTarget(t *testing.T) {
|
||||
"an inactive target is skipped, not an error",
|
||||
)
|
||||
|
||||
tasks := notif.Tasks()
|
||||
require.Len(t, tasks, 1)
|
||||
assert.Equal(t, active.ID, tasks[0].TargetID)
|
||||
|
||||
webhookDB, err := dbMgr.GetDB(wh.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
fresh := theOtherEvent(t, listEvents(t, webhookDB), original.ID)
|
||||
require.Len(t, listDeliveries(t, webhookDB, fresh.ID), 1)
|
||||
deliveries := listDeliveries(t, webhookDB, fresh.ID)
|
||||
require.Len(t, deliveries, 1)
|
||||
assert.Equal(t, active.ID, deliveries[0].TargetID)
|
||||
}
|
||||
|
||||
// TestHandleEventResubmit_NoActiveTargetsStillStoresEvent proves a
|
||||
// source with nothing to deliver to behaves as the receiver does: the
|
||||
// event is stored, nothing is queued, and the operator is told so
|
||||
// rather than being shown an error.
|
||||
// event is stored with no delivery, and the operator is told so rather
|
||||
// than being shown an error.
|
||||
func TestHandleEventResubmit_NoActiveTargetsStillStoresEvent(
|
||||
t *testing.T,
|
||||
) {
|
||||
@@ -473,10 +413,9 @@ func TestHandleEventResubmit_NoActiveTargetsStillStoresEvent(
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -493,12 +432,14 @@ func TestHandleEventResubmit_NoActiveTargetsStillStoresEvent(
|
||||
w.Header().Get("Location"),
|
||||
)
|
||||
|
||||
assert.Empty(t, notif.Tasks())
|
||||
|
||||
webhookDB, err := dbMgr.GetDB(wh.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Len(t, listEvents(t, webhookDB), 2)
|
||||
events := listEvents(t, webhookDB)
|
||||
assert.Len(t, events, 2)
|
||||
|
||||
fresh := theOtherEvent(t, events, original.ID)
|
||||
assert.Empty(t, listDeliveries(t, webhookDB, fresh.ID))
|
||||
}
|
||||
|
||||
// TestHandleEventResubmit_RefusesEventOfAnotherWebhook proves the
|
||||
@@ -514,10 +455,9 @@ func TestHandleEventResubmit_RefusesEventOfAnotherWebhook(
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
notif *recordingNotifier
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if)
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -548,10 +488,49 @@ func TestHandleEventResubmit_RefusesEventOfAnotherWebhook(
|
||||
w = postResubmit(t, h, sess, mine.ID, "not-a-uuid")
|
||||
assert.Equal(t, http.StatusNotFound, w.Code)
|
||||
|
||||
assert.Empty(
|
||||
t, notif.Tasks(),
|
||||
"a refused resubmit must queue nothing",
|
||||
for _, webhookID := range []string{theirs.ID, mine.ID} {
|
||||
webhookDB, err := dbMgr.GetDB(webhookID)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Len(
|
||||
t, listEvents(t, webhookDB), 1,
|
||||
"a refused resubmit must store nothing",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandleEventResubmit_FailedStoreIsAnError proves a copy the engine
|
||||
// cannot store answers with the 500 error page and leaves no new event.
|
||||
func TestHandleEventResubmit_FailedStoreIsAnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
h *handlers.Handlers
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
|
||||
wh := seedWebhook(t, db)
|
||||
original := seedStoredEvent(t, dbMgr, wh.ID, `{"not":"stored"}`)
|
||||
|
||||
// Without its event totals table the webhook's database refuses
|
||||
// the totals write, which comes after the event row.
|
||||
webhookDB, err := dbMgr.GetDB(wh.ID)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t,
|
||||
webhookDB.Migrator().DropTable(&database.EventTotals{}))
|
||||
|
||||
w := postResubmit(t, h, sess, wh.ID, original.ID)
|
||||
|
||||
assert.Equal(t, http.StatusInternalServerError, w.Code)
|
||||
assert.Contains(t, w.Body.String(), "500 Internal Server Error")
|
||||
assert.Len(t, listEvents(t, webhookDB), 1)
|
||||
}
|
||||
|
||||
// TestHandleSourceLogs_ShowsResubmitProvenance proves the event log
|
||||
|
||||
@@ -64,7 +64,7 @@ type HandlersParams struct {
|
||||
Healthcheck *healthcheck.Healthcheck
|
||||
Session *session.Session
|
||||
Middleware *middleware.Middleware
|
||||
Notifier delivery.Notifier
|
||||
Engine *delivery.Engine
|
||||
Archives archive.Archives
|
||||
CircuitBreakers delivery.CircuitBreakers
|
||||
SSRFGuard *delivery.Guard
|
||||
@@ -82,7 +82,7 @@ type Handlers struct {
|
||||
dbMgr *database.WebhookDBManager
|
||||
session *session.Session
|
||||
mw *middleware.Middleware
|
||||
notifier delivery.Notifier
|
||||
engine *delivery.Engine
|
||||
archives archive.Archives
|
||||
breakers delivery.CircuitBreakers
|
||||
mtr *metrics.Set
|
||||
@@ -138,7 +138,7 @@ func New(
|
||||
s.dbMgr = params.WebhookDBMgr
|
||||
s.session = params.Session
|
||||
s.mw = params.Middleware
|
||||
s.notifier = params.Notifier
|
||||
s.engine = params.Engine
|
||||
s.archives = params.Archives
|
||||
s.breakers = params.CircuitBreakers
|
||||
s.mtr = params.Metrics
|
||||
|
||||
@@ -30,33 +30,6 @@ import (
|
||||
"sneak.berlin/go/webhooker/internal/session"
|
||||
)
|
||||
|
||||
// recordingNotifier is a delivery.Notifier that records the tasks it
|
||||
// was handed, so a test can prove a handler queued the delivery it
|
||||
// claims to have queued — and, on the refusal paths, that it queued
|
||||
// nothing.
|
||||
type recordingNotifier struct {
|
||||
mu sync.Mutex
|
||||
tasks []delivery.Task
|
||||
}
|
||||
|
||||
func (n *recordingNotifier) Notify(tasks []delivery.Task) {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
|
||||
n.tasks = append(n.tasks, tasks...)
|
||||
}
|
||||
|
||||
// Tasks returns a copy of the recorded tasks.
|
||||
func (n *recordingNotifier) Tasks() []delivery.Task {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
|
||||
out := make([]delivery.Task, len(n.tasks))
|
||||
copy(out, n.tasks)
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
// recordingArchives is an archive.Archives that records what it
|
||||
// was asked to do, so a test can prove that a deletion or rename
|
||||
// path reached the archive writers. After FailRenames, every
|
||||
@@ -264,11 +237,12 @@ func newTestAppWithConfig(
|
||||
database.NewWebhookDBManager,
|
||||
healthcheck.New,
|
||||
session.New,
|
||||
func() *recordingNotifier {
|
||||
return &recordingNotifier{}
|
||||
},
|
||||
func(n *recordingNotifier) delivery.Notifier {
|
||||
return n
|
||||
archive.NewWriters,
|
||||
// The engine is built on a lifecycle that is never started,
|
||||
// so no worker runs: every delivery the engine creates for a
|
||||
// handler stays pending, where the test can read it.
|
||||
func(p delivery.EngineParams) *delivery.Engine {
|
||||
return delivery.New(fxtest.NewLifecycle(t), p)
|
||||
},
|
||||
func() *recordingArchives {
|
||||
return &recordingArchives{}
|
||||
|
||||
+20
-264
@@ -2,14 +2,11 @@ package handlers
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/delivery"
|
||||
"sneak.berlin/go/webhooker/internal/middleware"
|
||||
)
|
||||
|
||||
@@ -70,8 +67,9 @@ func (h *Handlers) HandleWebhook() http.HandlerFunc {
|
||||
}
|
||||
}
|
||||
|
||||
// processWebhookRequest reads the body, serializes headers, loads
|
||||
// targets, and delivers the event.
|
||||
// processWebhookRequest reads the request into an event and has the
|
||||
// delivery engine store it, with a delivery for each of the webhook's
|
||||
// active targets, and answers the sender.
|
||||
//
|
||||
// Nothing about the request itself is authenticated: the entrypoint
|
||||
// UUID in the path is the credential, and reaching here means it
|
||||
@@ -93,30 +91,24 @@ func (h *Handlers) processWebhookRequest(
|
||||
return
|
||||
}
|
||||
|
||||
targets, err := h.loadActiveTargets(entrypoint.WebhookID)
|
||||
event := &database.Event{
|
||||
WebhookID: entrypoint.WebhookID,
|
||||
EntrypointID: entrypoint.ID,
|
||||
Method: r.Method,
|
||||
RawQuery: r.URL.RawQuery,
|
||||
Headers: string(headersJSON),
|
||||
ContentType: r.Header.Get("Content-Type"),
|
||||
Body: string(body),
|
||||
}
|
||||
|
||||
deliveries, err := h.engine.StoreEvent(event)
|
||||
if err != nil {
|
||||
h.receiverError(w, "failed to query targets", err)
|
||||
h.receiverError(w, "failed to store webhook event", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
h.createAndDeliverEvent(
|
||||
w, r, entrypoint, body, headersJSON, targets,
|
||||
)
|
||||
}
|
||||
|
||||
// loadActiveTargets returns all active targets for a webhook.
|
||||
func (h *Handlers) loadActiveTargets(
|
||||
webhookID string,
|
||||
) ([]database.Target, error) {
|
||||
var targets []database.Target
|
||||
|
||||
err := h.db.DB().Where(
|
||||
"webhook_id = ? AND active = ?",
|
||||
webhookID, true,
|
||||
).Find(&targets).Error
|
||||
|
||||
return targets, err
|
||||
h.finishWebhookResponse(w, event, entrypoint, deliveries)
|
||||
}
|
||||
|
||||
// lookupEntrypoint finds an entrypoint by UUID path.
|
||||
@@ -177,28 +169,6 @@ func (h *Handlers) readWebhookBody(
|
||||
return body, true
|
||||
}
|
||||
|
||||
// createAndDeliverEvent stores the received event, fans it out to the
|
||||
// webhook's targets, and answers the sender.
|
||||
func (h *Handlers) createAndDeliverEvent(
|
||||
w http.ResponseWriter,
|
||||
r *http.Request,
|
||||
entrypoint database.Entrypoint,
|
||||
body, headersJSON []byte,
|
||||
targets []database.Target,
|
||||
) {
|
||||
event, tasks, err := h.createAndFanOut(
|
||||
requestEventSource(r, entrypoint, headersJSON, body),
|
||||
targets,
|
||||
)
|
||||
if err != nil {
|
||||
h.receiverError(w, "failed to store webhook event", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
h.finishWebhookResponse(w, event, entrypoint, tasks)
|
||||
}
|
||||
|
||||
// receiverError logs an error and answers the sender with a plain-text
|
||||
// 500. The receiver's answers are for programs, so it never sends the
|
||||
// error page the web UI uses.
|
||||
@@ -212,172 +182,19 @@ func (h *Handlers) receiverError(
|
||||
)
|
||||
}
|
||||
|
||||
// eventSource carries the fields a new event is built from. The
|
||||
// receiver fills it from the live request; the resubmit handler fills
|
||||
// it from a stored event. Both then go through createAndFanOut, so an
|
||||
// event is constructed and fanned out in one place however it entered
|
||||
// the system.
|
||||
type eventSource struct {
|
||||
WebhookID string
|
||||
EntrypointID string
|
||||
Method string
|
||||
RawQuery string
|
||||
HeadersJSON string
|
||||
ContentType string
|
||||
Body []byte
|
||||
|
||||
// ResubmittedFromID names the event this one copies. Only the
|
||||
// resubmit path sets it.
|
||||
ResubmittedFromID *string
|
||||
}
|
||||
|
||||
// event builds the row this source stores.
|
||||
func (s eventSource) event() *database.Event {
|
||||
return &database.Event{
|
||||
WebhookID: s.WebhookID,
|
||||
EntrypointID: s.EntrypointID,
|
||||
Method: s.Method,
|
||||
RawQuery: s.RawQuery,
|
||||
Headers: s.HeadersJSON,
|
||||
Body: string(s.Body),
|
||||
BodyBytes: int64(len(s.Body)),
|
||||
ContentType: s.ContentType,
|
||||
ResubmittedFromID: s.ResubmittedFromID,
|
||||
}
|
||||
}
|
||||
|
||||
// requestEventSource describes the event a live receiver request
|
||||
// stores.
|
||||
func requestEventSource(
|
||||
r *http.Request,
|
||||
entrypoint database.Entrypoint,
|
||||
headersJSON, body []byte,
|
||||
) eventSource {
|
||||
return eventSource{
|
||||
WebhookID: entrypoint.WebhookID,
|
||||
EntrypointID: entrypoint.ID,
|
||||
Method: r.Method,
|
||||
RawQuery: r.URL.RawQuery,
|
||||
HeadersJSON: string(headersJSON),
|
||||
ContentType: r.Header.Get("Content-Type"),
|
||||
Body: body,
|
||||
}
|
||||
}
|
||||
|
||||
// createAndFanOut writes the event and one pending delivery per target,
|
||||
// and adds them to the webhook's running totals, in a single
|
||||
// transaction, then hands the tasks to the delivery engine. Every
|
||||
// event is created here, received or resubmitted, so a resubmitted
|
||||
// event is retried, SSRF-guarded and circuit-broken exactly as a
|
||||
// received one is. Per-delivery replay is the one other path that
|
||||
// creates a delivery: it adds one to an existing event without
|
||||
// coming through here.
|
||||
//
|
||||
// The tasks are returned as well as queued, so a caller can report how
|
||||
// many targets the event went to.
|
||||
func (h *Handlers) createAndFanOut(
|
||||
src eventSource,
|
||||
targets []database.Target,
|
||||
) (*database.Event, []delivery.Task, error) {
|
||||
webhookDB, err := h.dbMgr.GetDB(src.WebhookID)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf(
|
||||
"getting webhook database: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
tx := webhookDB.Begin()
|
||||
if tx.Error != nil {
|
||||
return nil, nil, fmt.Errorf(
|
||||
"beginning transaction: %w", tx.Error,
|
||||
)
|
||||
}
|
||||
|
||||
event := src.event()
|
||||
|
||||
err = tx.Create(event).Error
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
|
||||
return nil, nil, fmt.Errorf("creating event: %w", err)
|
||||
}
|
||||
|
||||
tasks, err := buildDeliveryTasks(
|
||||
tx, event, targets, inlineBody(src.Body),
|
||||
)
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
err = database.AddEventTotals(tx, database.EventTotals{
|
||||
Events: 1, LastEventAt: &event.CreatedAt,
|
||||
})
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
// A resubmitted copy did not arrive on its entrypoint's URL, so it
|
||||
// leaves the entrypoint's last event as it is.
|
||||
if src.ResubmittedFromID == nil {
|
||||
err = database.AddEntrypointTotals(tx, database.EntrypointTotals{
|
||||
EntrypointID: event.EntrypointID, LastEventAt: event.CreatedAt,
|
||||
})
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
err = tx.Commit().Error
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf(
|
||||
"committing transaction: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
// Counted here, after the commit: an event exists once it is
|
||||
// durably stored, which is what the delivery counters are
|
||||
// compared against on a dashboard. A resubmitted event counts
|
||||
// too, because it produces deliveries that the delivery side
|
||||
// counts; the resubmit counter is what separates the two.
|
||||
h.mtr.EventReceived()
|
||||
|
||||
if len(tasks) > 0 {
|
||||
h.notifier.Notify(tasks)
|
||||
}
|
||||
|
||||
return event, tasks, nil
|
||||
}
|
||||
|
||||
// inlineBody returns a pointer to body as a string if it fits
|
||||
// within the inline size limit, or nil otherwise.
|
||||
func inlineBody(body []byte) *string {
|
||||
if len(body) < delivery.MaxInlineBodySize {
|
||||
s := string(body)
|
||||
|
||||
return &s
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// finishWebhookResponse logs the event and writes the HTTP response.
|
||||
// finishWebhookResponse logs the stored event, with how many
|
||||
// deliveries it has, and writes the HTTP response.
|
||||
func (h *Handlers) finishWebhookResponse(
|
||||
w http.ResponseWriter,
|
||||
event *database.Event,
|
||||
entrypoint database.Entrypoint,
|
||||
tasks []delivery.Task,
|
||||
deliveries int,
|
||||
) {
|
||||
h.log.Info("webhook event created",
|
||||
"event_id", event.ID,
|
||||
"webhook_id", entrypoint.WebhookID,
|
||||
"entrypoint_id", entrypoint.ID,
|
||||
"target_count", len(tasks),
|
||||
"target_count", deliveries,
|
||||
)
|
||||
|
||||
w.WriteHeader(http.StatusOK)
|
||||
@@ -389,64 +206,3 @@ func (h *Handlers) finishWebhookResponse(
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// buildDeliveryTasks creates one pending delivery per target in the
|
||||
// transaction, adds each to its target's totals, and returns the tasks
|
||||
// for the delivery engine. The caller owns the transaction and rolls
|
||||
// it back on error.
|
||||
func buildDeliveryTasks(
|
||||
tx *gorm.DB,
|
||||
event *database.Event,
|
||||
targets []database.Target,
|
||||
bodyPtr *string,
|
||||
) ([]delivery.Task, error) {
|
||||
tasks := make([]delivery.Task, 0, len(targets))
|
||||
|
||||
for i := range targets {
|
||||
dlv := &database.Delivery{
|
||||
EventID: event.ID,
|
||||
TargetID: targets[i].ID,
|
||||
Status: database.DeliveryStatusPending,
|
||||
}
|
||||
|
||||
err := tx.Create(dlv).Error
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"creating delivery for target %s: %w",
|
||||
targets[i].ID, err,
|
||||
)
|
||||
}
|
||||
|
||||
err = database.AddTargetTotals(tx, database.TargetTotals{
|
||||
TargetID: targets[i].ID, Deliveries: 1,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tasks = append(tasks, delivery.Task{
|
||||
DeliveryID: dlv.ID,
|
||||
EventID: event.ID,
|
||||
WebhookID: event.WebhookID,
|
||||
EntrypointID: event.EntrypointID,
|
||||
TargetID: targets[i].ID,
|
||||
TargetName: targets[i].Name,
|
||||
TargetType: targets[i].Type,
|
||||
TargetURL: targets[i].URL,
|
||||
TargetHeaders: targets[i].Headers,
|
||||
TimeoutSeconds: targets[i].TimeoutSeconds,
|
||||
ForwardQuery: targets[i].ForwardQuery,
|
||||
ArchiveExpiry: targets[i].ArchiveExpiry,
|
||||
ArchiveRotation: targets[i].ArchiveRotation,
|
||||
MaxAttempts: targets[i].MaxAttempts,
|
||||
Method: event.Method,
|
||||
RawQuery: event.RawQuery,
|
||||
Headers: event.Headers,
|
||||
ContentType: event.ContentType,
|
||||
Body: bodyPtr,
|
||||
AttemptNum: 1,
|
||||
})
|
||||
}
|
||||
|
||||
return tasks, nil
|
||||
}
|
||||
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"net/netip"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/google/uuid"
|
||||
@@ -19,7 +18,6 @@ import (
|
||||
"gorm.io/gorm/clause"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/delivery"
|
||||
"sneak.berlin/go/webhooker/internal/handlers"
|
||||
"sneak.berlin/go/webhooker/internal/middleware"
|
||||
"sneak.berlin/go/webhooker/internal/middleware/middlewaretest"
|
||||
@@ -209,50 +207,19 @@ func receivedLine(t *testing.T, buf *bytes.Buffer) map[string]any {
|
||||
return found[0]
|
||||
}
|
||||
|
||||
// targetWithEverySetting returns target settings with all six set, the
|
||||
// archive ones included, so a task that drops one of them differs from
|
||||
// the target.
|
||||
func targetWithEverySetting() database.Target {
|
||||
return database.Target{
|
||||
URL: replayTargetURL,
|
||||
Headers: map[string]string{"X-Api-Key": "k"},
|
||||
TimeoutSeconds: 7,
|
||||
ForwardQuery: true,
|
||||
ArchiveExpiry: time.Hour,
|
||||
ArchiveRotation: "daily",
|
||||
}
|
||||
}
|
||||
|
||||
// assertTaskCarriesTargetSettings proves a delivery task carries every
|
||||
// one of the target's settings. They are copied into the task one field
|
||||
// at a time, and a field left out is sent empty without any error.
|
||||
func assertTaskCarriesTargetSettings(
|
||||
t *testing.T, task delivery.Task, target *database.Target,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
assert.Equal(t, target.ID, task.TargetID)
|
||||
assert.Equal(t, target.URL, task.TargetURL)
|
||||
assert.Equal(t, target.Headers, task.TargetHeaders)
|
||||
assert.Equal(t, target.TimeoutSeconds, task.TimeoutSeconds)
|
||||
assert.Equal(t, target.ForwardQuery, task.ForwardQuery)
|
||||
assert.Equal(t, target.ArchiveExpiry, task.ArchiveExpiry)
|
||||
assert.Equal(t, target.ArchiveRotation, task.ArchiveRotation)
|
||||
}
|
||||
|
||||
// TestHandleWebhook_TaskCarriesEveryTargetSetting proves the task a
|
||||
// received event hands the delivery engine carries every one of the
|
||||
// target's settings.
|
||||
func TestHandleWebhook_TaskCarriesEveryTargetSetting(t *testing.T) {
|
||||
// TestHandleWebhook_StoresTheEvent proves an accepted request is
|
||||
// answered with the receiver's 200 and body, and that the event is
|
||||
// stored with a pending delivery to the webhook's active target.
|
||||
func TestHandleWebhook_StoresTheEvent(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
h *handlers.Handlers
|
||||
db *database.Database
|
||||
notif *recordingNotifier
|
||||
dbMgr *database.WebhookDBManager
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &db, ¬if)
|
||||
app := newTestApp(t, &h, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -261,7 +228,7 @@ func TestHandleWebhook_TaskCarriesEveryTargetSetting(t *testing.T) {
|
||||
entrypoint := seedEntrypoint(t, db, webhook.ID)
|
||||
target := seedConfiguredTarget(
|
||||
t, db, webhook.ID, database.TargetTypeHTTP,
|
||||
targetWithEverySetting(),
|
||||
database.Target{URL: replayTargetURL},
|
||||
)
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
@@ -271,8 +238,55 @@ func TestHandleWebhook_TaskCarriesEveryTargetSetting(t *testing.T) {
|
||||
))
|
||||
|
||||
require.Equal(t, http.StatusOK, w.Code)
|
||||
assert.JSONEq(t, `{"status":"ok"}`, w.Body.String())
|
||||
|
||||
tasks := notif.Tasks()
|
||||
require.Len(t, tasks, 1)
|
||||
assertTaskCarriesTargetSettings(t, tasks[0], target)
|
||||
webhookDB, err := dbMgr.GetDB(webhook.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
events := listEvents(t, webhookDB)
|
||||
require.Len(t, events, 1)
|
||||
assert.Equal(t, entrypoint.ID, events[0].EntrypointID)
|
||||
|
||||
deliveries := listDeliveries(t, webhookDB, events[0].ID)
|
||||
require.Len(t, deliveries, 1)
|
||||
assert.Equal(t, target.ID, deliveries[0].TargetID)
|
||||
assert.Equal(t, database.DeliveryStatusPending, deliveries[0].Status)
|
||||
}
|
||||
|
||||
// TestHandleWebhook_FailedStoreAnswersPlainError proves that an event
|
||||
// the engine cannot store is answered with the receiver's plain-text
|
||||
// 500, and that nothing of it is left stored.
|
||||
func TestHandleWebhook_FailedStoreAnswersPlainError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
h *handlers.Handlers
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
|
||||
webhook := seedWebhook(t, db)
|
||||
entrypoint := seedEntrypoint(t, db, webhook.ID)
|
||||
|
||||
// Without its event totals table the webhook's database refuses
|
||||
// the totals write, which comes after the event row.
|
||||
webhookDB, err := dbMgr.GetDB(webhook.ID)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t,
|
||||
webhookDB.Migrator().DropTable(&database.EventTotals{}))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
h.HandleWebhook().ServeHTTP(w, postRequest(
|
||||
"/h/"+entrypoint.Path, nil,
|
||||
map[string]string{"uuid": entrypoint.Path},
|
||||
))
|
||||
|
||||
assert.Equal(t, http.StatusInternalServerError, w.Code)
|
||||
assert.Equal(t, "Internal server error\n", w.Body.String())
|
||||
assert.Empty(t, listEvents(t, webhookDB))
|
||||
}
|
||||
|
||||
@@ -21,8 +21,9 @@ const budget = 64
|
||||
|
||||
// batchRunes is how many consecutive code points the charge test logs
|
||||
// in one value from U+1000 up. Logging each of those on its own line
|
||||
// is too slow for the suite under the race detector; 4,096 at a time
|
||||
// is 271 batches, each logged on two lines, so 542 lines per handler.
|
||||
// was too slow for the suite when it ran under the race detector;
|
||||
// 4,096 at a time is 271 batches, each logged on two lines, so 542
|
||||
// lines per handler.
|
||||
const batchRunes = 4096
|
||||
|
||||
// newHandlers are the two handlers internal/logger can install. Time
|
||||
@@ -152,10 +153,10 @@ func TestEncodedBytes_ChargesAtLeastWhatTheHandlersEmit(t *testing.T) {
|
||||
batches = append(batches, batch.String())
|
||||
}
|
||||
|
||||
// What EncodedBytes charges for each batch. Under -race -cover this
|
||||
// takes longer than logging the batches, so it is worked out once,
|
||||
// by whichever handler finishes logging first, while the other is
|
||||
// still logging.
|
||||
// What EncodedBytes charges for each batch. When the suite ran
|
||||
// under -race -cover this took longer than logging the batches, so
|
||||
// it is worked out once, by whichever handler finishes logging
|
||||
// first, while the other is still logging.
|
||||
charged := sync.OnceValue(func() []int {
|
||||
costs := make([]int, len(batches))
|
||||
|
||||
|
||||
@@ -131,10 +131,6 @@ func run(
|
||||
return code, stdout.String(), stderr.String()
|
||||
}
|
||||
|
||||
type noopNotifier struct{}
|
||||
|
||||
func (n *noopNotifier) Notify([]delivery.Task) {}
|
||||
|
||||
type noopArchives struct {
|
||||
renameMu sync.Mutex
|
||||
}
|
||||
@@ -190,7 +186,12 @@ func newServerApp(
|
||||
database.NewWebhookDBManager,
|
||||
healthcheck.New,
|
||||
session.New,
|
||||
func() delivery.Notifier { return &noopNotifier{} },
|
||||
archive.NewWriters,
|
||||
// The engine is built on a lifecycle that is never started,
|
||||
// so no worker runs.
|
||||
func(p delivery.EngineParams) *delivery.Engine {
|
||||
return delivery.New(fxtest.NewLifecycle(t), p)
|
||||
},
|
||||
func() archive.Archives { return &noopArchives{} },
|
||||
func() delivery.CircuitBreakers {
|
||||
return &noopCircuitBreakers{}
|
||||
|
||||
@@ -10,14 +10,12 @@ import (
|
||||
"sneak.berlin/go/webhooker/internal/server"
|
||||
)
|
||||
|
||||
// earlyStopIterations is how many start/stop cycles the race test
|
||||
// earlyStopIterations is how many start/stop cycles the test below
|
||||
// runs. The window it aims at is the gap between the OnStart hook
|
||||
// returning and the serving goroutine reaching its first field
|
||||
// access, which is microseconds wide. The race detector reports an
|
||||
// unsynchronised pair whenever it observes one, but it has to observe
|
||||
// one, so a single cycle can miss purely on scheduling. Repetition
|
||||
// makes the observation reliable; the collaborators are built once,
|
||||
// so the cycles themselves are cheap.
|
||||
// access, which is microseconds wide, so a single cycle can miss it
|
||||
// purely on scheduling. Repetition makes hitting it reliable; the
|
||||
// collaborators are built once, so the cycles themselves are cheap.
|
||||
const earlyStopIterations = 25
|
||||
|
||||
// TestEarlyShutdown_NoPanicAndNoRace stops the application
|
||||
@@ -38,8 +36,9 @@ const earlyStopIterations = 25
|
||||
//
|
||||
// httpServer is now built in New, on the constructing goroutine, so
|
||||
// it is written before any hook exists and can never be nil.
|
||||
// sentryEnabled is atomic. This test is what catches either one
|
||||
// coming back — under -race, which is how the suite runs.
|
||||
// sentryEnabled is atomic. This test catches the nil dereference
|
||||
// coming back, as a panic. The suite runs without Go's race
|
||||
// detector, so the data race coming back would go unreported here.
|
||||
func TestEarlyShutdown_NoPanicAndNoRace(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -48,10 +48,6 @@ const (
|
||||
metricsAuthValue = "s3cret"
|
||||
)
|
||||
|
||||
type noopNotifier struct{}
|
||||
|
||||
func (n *noopNotifier) Notify([]delivery.Task) {}
|
||||
|
||||
// noopArchives satisfies handlers.New's archive.Archives
|
||||
// dependency. No test here checks what gets evicted or renamed, so
|
||||
// it records nothing. Its rename lock is a plain mutex.
|
||||
@@ -145,7 +141,12 @@ func newTestEnvWithConfig(
|
||||
database.NewWebhookDBManager,
|
||||
healthcheck.New,
|
||||
session.New,
|
||||
func() delivery.Notifier { return &noopNotifier{} },
|
||||
archive.NewWriters,
|
||||
// The engine is built on a lifecycle that is never started,
|
||||
// so no worker runs and nothing is sent anywhere.
|
||||
func(p delivery.EngineParams) *delivery.Engine {
|
||||
return delivery.New(fxtest.NewLifecycle(t), p)
|
||||
},
|
||||
func() archive.Archives { return &noopArchives{} },
|
||||
func() delivery.CircuitBreakers {
|
||||
return &noopCircuitBreakers{}
|
||||
|
||||
Reference in New Issue
Block a user