8 Commits
Author SHA1 Message Date
clawbot eda8e80844 Fix six README sentences (closes #565)
check / check (push) Canceled after 0s
Rewrite the six sentences the documents check failed, each checked
against the code. The Sign-in limits section now opens with its rule.
The recovery routine's sentences say it resumes a delivery: it queues a
pending one at once, but gives a retrying one the rest of its wait.

Model: opus-5-5
2026-10-26 06:53:59 +01:00
clawbot e584718eb5 Commit Alpine.js to the repo instead of unpacking it at build time (closes #163)
check / check (push) Canceled after 0s
static/js/alpine.min.js is now committed: package/dist/cdn.min.js from
the @alpinejs/csp 3.14.9 npm package, byte for byte, taken once from the
package file that was in 3p/. The build no longer unpacks it, so
script/assets, the assets make target, its Dockerfile steps and the
ignore entries for the file are gone, and so is the package file.
Prettier and ESLint both ignore the file, so it stays as published.
Comments and the README now describe the committed file.

Model: opus-5-5
2026-10-26 05:16:05 +01:00
clawbot a8d226942e Rewrite TODO.md sentence by sentence (closes #558)
check / check (push) Canceled after 0s
Rewritten from scratch, sections kept, every statement checked against the code. Passed independent review in round 4.

Model: fable-5-1 (writing); opus-5-5 (review, merge)
2026-10-25 12:05:48 +01:00
clawbot 6df1fc7875 Rewrite README.md sentence by sentence (#561)
check / check (push) Canceled after 0s
Rewritten from scratch, about 4,600 words, every statement checked against the code. Part of #558; passed independent review in round 10.

Model: fable-5-1 (writing); opus-5-5 (review, merge)
2026-10-25 12:05:41 +01:00
sneak 54b643044f Merge branch 'main' into next
check / check (push) Canceled after 0s
2026-10-25 02:51:56 +01:00
clawbot 55ccb28270 Run the tests with cgo off and without the race detector (closes #538)
check / check (push) Canceled after 0s
sneak ruled the race detector out
(#538 (comment)), and on
Linux it was the only reason the tests needed cgo. The test phase and the
browser test binary now build with CGO_ENABLED=0 and without -race, so no
test binary compiles mattn's C SQLite. The test phase keeps the builder
stage's Debian Go image. Its comments, the README and the code comments that
said the suite runs under -race now say what is true, and the README notes
the exception to the vendored REPO_POLICIES.md.

Model: opus-5-5
2026-10-21 17:28:50 +02:00
clawbot 06be075b70 Store events and create deliveries in the delivery engine (closes #544)
check / check (push) Canceled after 0s
The receiver, resubmit and replay handlers now read and check the
request and make one call. Engine.StoreEvent stores an event with a
pending delivery for each active target and the totals in one
transaction, then queues them; Engine.CreateReplay does the same for
one replay delivery. The handlers take *delivery.Engine, so the
Notifier interface and its bridge are gone, and the engine's notify is
no longer exported. Archives and CircuitBreakers stay because the
handler tests fake them. The tests that read a queued task moved to
the engine's tests; the handler tests use an engine that is never
started.

Model: opus-5-5
2026-10-16 00:51:26 +02:00
sneak 766f33a956 Milestone: next into main (#380)
check / check (push) Canceled after 0s
Reviewed-on: #380
2026-10-15 11:36:37 +02:00
30 changed files with 830 additions and 3618 deletions
+1 -3
View File
@@ -72,10 +72,8 @@
**/*.sublime-* **/*.sublime-*
# This repository's own host-built artifacts: the binary `make build` # This repository's own host-built artifacts: the binary `make build`
# writes, and the Alpine.js file `make assets` extracts from 3p/ (the # writes.
# build extracts its own).
/bin /bin
/static/js/alpine.min.js
# SQLite databases, which hold the session key and webhook payloads, at # SQLite databases, which hold the session key and webhook payloads, at
# any depth. # any depth.
-4
View File
@@ -77,7 +77,3 @@ data/
# Temporary files # Temporary files
tmp/ tmp/
temp/ temp/
# Alpine.js, extracted by `make assets` from its tarball in 3p/, which is
# what is committed.
/static/js/alpine.min.js
+1
View File
@@ -1,2 +1,3 @@
node_modules/ node_modules/
yarn.lock yarn.lock
static/js/alpine.min.js
Binary file not shown.
+15 -19
View File
@@ -92,10 +92,6 @@ COPY . .
# fmt-check` is the markdown-check stage above. # fmt-check` is the markdown-check stage above.
RUN if [ -n "$(gofmt -s -l .)" ]; then echo "gofmt needed on:"; gofmt -s -l .; exit 1; fi RUN if [ -n "$(gofmt -s -l .)" ]; then echo "gofmt needed on:"; gofmt -s -l .; exit 1; fi
# static/static.go embeds the Alpine.js file this extracts from 3p/; without
# it the static package does not compile and cannot be linted.
RUN script/assets
# The golangci-lint steps run with --network=none. `golangci-lint config # The golangci-lint steps run with --network=none. `golangci-lint config
# verify` is documented as fetching its JSON schema over HTTPS; this pinned # verify` is documented as fetching its JSON schema over HTTPS; this pinned
# image resolves the schema without any network, and --network=none enforces # image resolves the schema without any network, and --network=none enforces
@@ -111,11 +107,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. # Nothing is wanted from js-lint; the copy is what makes this phase run it.
COPY --from=js-lint /src/yarn.lock /dev/null COPY --from=js-lint /src/yarn.lock /dev/null
# Test phase. -race needs cgo and so a C compiler, which the Debian Go image # Test phase. The tests run with cgo off and so without Go's race detector,
# ships and the alpine one does not. `make test` (script/test) builds this # which needs cgo on Linux; sneak ruled the race detector out in
# stage alone; the build stage below depends on it. # 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 FROM golang:1.26.1-bookworm@sha256:4465644228bc2857a954b092167e12aa59c006a3492282a6c820bf4755fd64a4 AS test
WORKDIR /src WORKDIR /src
@@ -125,14 +126,11 @@ RUN go mod download
COPY . . COPY . .
# static/static.go embeds the Alpine.js file this extracts from 3p/.
RUN script/assets
# -timeout applies to each package on its own, so 90s has only to clear the # -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 # 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 # 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, # tests. The defaults are one per core, so without these the memory the run
# so the defaults (one per core) add up to several GB on a many-core host. # needs grows with the host's core count.
# #
# The first run has no -v: go test then prints one result line per package, # 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 # with its coverage, and for a package that fails, everything its tests
@@ -154,12 +152,12 @@ RUN script/assets
SHELL ["/bin/bash", "-o", "pipefail", "-c"] SHELL ["/bin/bash", "-o", "pipefail", "-c"]
RUN --mount=type=tmpfs,target=/tmp/tests,size=512m \ RUN --mount=type=tmpfs,target=/tmp/tests,size=512m \
export TMPDIR=/tmp/tests GOTMPDIR=/tmp; \ 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 '|' -)"; \ tests="$(awk '/^--- FAIL: / { print $3 }' /tmp/go-test.log | paste -s -d '|' -)"; \
packages="$(awk '/^FAIL\t/ { print $2 }' /tmp/go-test.log)"; \ packages="$(awk '/^FAIL\t/ { print $2 }' /tmp/go-test.log)"; \
if [ -n "$tests" ]; then \ if [ -n "$tests" ]; then \
echo "--- Rerunning the failed tests with -v for details ---"; \ 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; \ fi; \
exit 1 exit 1
@@ -206,10 +204,8 @@ RUN version="$(make version VERSION="$VERSION")"; \
esac; \ esac; \
fi fi
# Builds through the Makefile's build target, which runs script/assets # The Makefile's build target builds with cgo off, so the binary is
# (Alpine.js, extracted from its tarball in 3p/) first. The target builds # statically linked and runs on the Alpine runtime image below.
# with cgo off, so the binary is statically linked and runs on the Alpine
# runtime image below as it is.
RUN make build VERSION="$VERSION" RUN make build VERSION="$VERSION"
# Runtime stage # Runtime stage
+3 -3
View File
@@ -15,9 +15,9 @@ RUN go mod download
COPY . . COPY . .
# The test binary embeds the templates and static files, so the browser # 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 # stage needs nothing else. It is built with cgo off and -p 4 keeps the
# the test phase of Dockerfile. # compile's memory down, both as in the test phase of Dockerfile.
RUN make assets && go test -c -p 4 -tags browser -o /browser.test ./internal/server RUN 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 # 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 # browser is on PATH as headless-shell, where the test's browser library
+3 -6
View File
@@ -1,4 +1,4 @@
.PHONY: bootstrap setup assets test test-browser lint fmt fmt-check check build run dev deps docker clean hooks css css-check version .PHONY: bootstrap setup test test-browser lint fmt fmt-check check build run dev deps docker clean hooks css css-check version
# Default target # Default target
.DEFAULT_GOAL := check .DEFAULT_GOAL := check
@@ -22,9 +22,6 @@ bootstrap:
setup: setup:
@script/setup @script/setup
assets:
@script/assets
test: test:
@script/test @script/test
@@ -47,13 +44,13 @@ check:
# REPO_POLICIES.md does: the binary carries no build paths, symbol table or # REPO_POLICIES.md does: the binary carries no build paths, symbol table or
# debug information, and on Linux, the image's included, it is statically # debug information, and on Linux, the image's included, it is statically
# linked. # linked.
build: assets build:
CGO_ENABLED=0 go build -trimpath -ldflags '-s -w -X main.version=$(VERSION)' -o bin/webhooker ./cmd/webhooker CGO_ENABLED=0 go build -trimpath -ldflags '-s -w -X main.version=$(VERSION)' -o bin/webhooker ./cmd/webhooker
run: build run: build
./bin/webhooker ./bin/webhooker
dev: assets dev:
go run ./cmd/webhooker go run ./cmd/webhooker
deps: deps:
+618 -3413
View File
File diff suppressed because it is too large Load Diff
+51 -46
View File
@@ -1,72 +1,77 @@
# Workflow # 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 - make sure a tracked issue exists with a definition of done
- branch from `next` (never from `main`) - branch from `next`, never from `main`
- do the work; open a PR based on `next` (never on `main`) - do the work and push the branch
- pass an independent review, then the manager squash-merges into `next` - open a PR with base `next`, never `main`
- push; nothing stays local-only - 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 The repository manager is the role that merges each issue's PR into `next`.
`main` without notice. One `next` -> `main` PR accumulates the milestone; Nothing stays local-only: every commit is pushed.
releases are cut from `main` separately.
The tracker (https://git.eeqj.de/sneak/webhooker/issues) is the record of what `next` is the branch for the next milestone. It must stay green and mergeable
is done and what is under way. This file keeps no list of either. into `main` at all times. A milestone's work reaches `main` through one PR from
`next`. The owner, @sneak, merges that PR. 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 # Status
What is still open for 1.0.0 is in its milestone 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 (https://git.eeqj.de/sneak/webhooker/milestone/9). `v1.0.0` is not tagged. Only
only @sneak tags it. @sneak tags it.
Delivery is at-least-once by design, not by accident: a send whose result row Delivery is at least once by design. When webhooker cannot record the outcome of
does not land is attempted again, so a receiver can see a duplicate. That is a delivery, it sends that delivery again. The target can then receive the same
deliberate — the alternative is a silent lost delivery — and the README says so event twice. The alternative is a lost delivery that nothing reports. The README
under Rationale. It is not a defect to re-file. says so. Such a duplicate is not a defect to re-file.
# Next Step # Next Step
@sneak runs webhooker in limited production, then tags `v1.0.0`. Merging `next` @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` into `main` is a separate act. It waits on neither the production run nor the
is kept mergeable at all times, which is the point of the branch. tag.
# Future Steps # Future Steps
- Per-webhook rate limits, set in the web UI and enforced in the receiver - Per-webhook rate limits, set in the web UI and enforced at the entrypoint.
handler, on top of the receiver's existing limits, which `RECEIVER_RATE_LIMIT` Today `RECEIVER_RATE_LIMIT` is the only rate-limit setting. It is how many
sets for the whole server (see Rate Limiting in the README) requests per minute one client address may send to one entrypoint. The same
- API key authentication for programmatic access (the `APIKey` model exists, but address is also capped at ten times that across all entrypoints together.
nothing creates a key and no middleware checks one) - API key authentication for programmatic access. The `APIKey` model and its
- REST API v1 `api_keys` table exist. Nothing creates a key and nothing checks one.
- CRUD for webhooks, entrypoints, targets - REST API v1. `/api/v1` is mounted with no routes today.
- CRUD for webhooks, entrypoints and targets
- event viewing and filtering endpoints - event viewing and filtering endpoints
- event redelivery endpoint - event redelivery endpoint
- OpenAPI specification - OpenAPI specification
- Analytics in the web UI: response times summarised per webhook or per target, - 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, and figures for all webhooks together. Today the webhook page shows event,
delivery and failure counts and failure percentages, the target list counts 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 each target's deliveries. The event log shows every attempt's duration.
attempt's duration, and `/metrics` carries server-wide counts and delivery `/metrics` carries server-wide counts and delivery durations.
durations - A remember-me option at login. Today every session survives a browser restart.
- A remember-me option at login. Every session already survives a browser It lasts up to 7 days. It ends sooner after `SESSION_IDLE_TIMEOUT` (default 24
restart: it lasts up to 7 days, and ends sooner after `SESSION_IDLE_TIMEOUT` hours) without use.
(default 24 hours) without activity - Password reset over the web for a forgotten password. Today a signed-in user
- Password reset flow for a forgotten password over the web. The authenticated can change their password. A lost password is reset from the command line with
password _change_ flow already landed, and a lost password is recoverable from `webhooker resetpw` (https://git.eeqj.de/sneak/webhooker/issues/208).
the console with `webhooker resetpw`
(https://git.eeqj.de/sneak/webhooker/issues/208)
- Later, nice to have - Later, nice to have
- email delivery target type - email delivery target type
- SNS and S3 delivery targets - SNS and S3 delivery targets
- data transformations (e.g. a configurable Slack message format; the Slack - data transformations, such as a configurable Slack message format. The
target sends one fixed format today) Slack target sends one fixed format today.
- JSONL file delivery with periodic S3 upload (the `database` target already - JSONL file delivery with periodic S3 upload. Today an archive target
archives events to SQLite files, rotated hourly, daily or monthly if set, writes events to SQLite files. An archive target can rotate those files
and the webhook page offers them as one gzipped JSON download) 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 - webhook event search, and filtering beyond the event log's failed and
pending views pending views
- multi-user with role-based access control (each webhook already belongs to - multi-user with role-based access control. Today each webhook belongs to
one account, but there is only the account made at first boot and no way one account. Only the account made at first boot exists. There is no way
to add another) to add another.
+1 -1
View File
@@ -5,7 +5,7 @@
// REPO_POLICIES.md states that a linter can check: const for everything, // REPO_POLICIES.md states that a linter can check: const for everything,
// let only for a variable that is reassigned, never var. // let only for a variable that is reassigned, never var.
export default [ export default [
// Alpine.js, extracted from 3p/ by make assets; not ours to lint. // Alpine.js, committed as published; not ours to lint.
{ ignores: ["static/js/alpine.min.js"] }, { ignores: ["static/js/alpine.min.js"] },
{ {
// The pages load static/js/app.js as a classic script, not a module. // The pages load static/js/app.js as a classic script, not a module.
+1 -1
View File
@@ -303,7 +303,7 @@ func (r *racingWrites) wait() (bool, error) {
// contending for the writer's mutex when the eviction takes it. // contending for the writer's mutex when the eviction takes it.
// The write that wins the mutex after the eviction must abandon // The write that wins the mutex after the eviction must abandon
// its work rather than reopen the archive, leaving the writer // its work rather than reopen the archive, leaving the writer
// permanently handle-free. Run under -race. // permanently handle-free.
func TestEvictWebhook_RacingWriteDoesNotReopenHandle( func TestEvictWebhook_RacingWriteDoesNotReopenHandle(
t *testing.T, t *testing.T,
) { ) {
+5 -3
View File
@@ -959,9 +959,11 @@ func TestArchiveSweep_SkipsDeletedWebhookTargets(t *testing.T) {
) )
} }
// TestArchiveSweep_ConcurrentWrites proves the sweep serialises // TestArchiveSweep_ConcurrentWrites runs the sweep while writes to
// against writes through the target's writer mutex. Run under // the same target are under way. The sweep serialises against them
// -race, an unsynchronised sweep would be caught here. // 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) { func TestArchiveSweep_ConcurrentWrites(t *testing.T) {
t.Parallel() t.Parallel()
+4 -3
View File
@@ -67,9 +67,10 @@ func DefaultPasswordConfig() *PasswordConfig {
// testArgon2Memory is the Argon2id memory cost, in KiB, that a test // testArgon2Memory is the Argon2id memory cost, in KiB, that a test
// binary hashes with: 1 MB instead of the shipped 64 MB. Every test // binary hashes with: 1 MB instead of the shipped 64 MB. Every test
// that starts a database hashes the bootstrap admin password, dozens // that starts a database hashes the bootstrap admin password, dozens
// of them run in parallel, and under the race detector each 64 MB hash // of them run in parallel, and each 64 MB hash holds at least 64 MB
// holds about 150 MB. VerifyPassword reads the cost from the hash it // while it runs (about 150 MB when the suite ran under the race
// checks, so verification follows. // detector). VerifyPassword reads the cost from the hash it checks, so
// verification follows.
const testArgon2Memory = 1024 const testArgon2Memory = 1024
// hashAtShippedCostInTest makes a test binary hash at the shipped // hashAtShippedCostInTest makes a test binary hash at the shipped
+45 -43
View File
@@ -23,10 +23,11 @@ import (
const ( const (
// deliveryChannelSize is the buffer size for the delivery // deliveryChannelSize is the buffer size for the delivery
// channel. The Tasks of newly stored deliveries are sent // channel. The Tasks of newly stored deliveries, and of the
// here. Workers drain this channel. Sized large enough // pending deliveries the recovery routine hands back, are
// that storing an event should never block under normal // sent here. Workers drain this channel. Sized large enough
// load. // that no task should be dropped under normal load: one that
// does not fit is left for the sweep.
deliveryChannelSize = 10000 deliveryChannelSize = 10000
// retryChannelSize is the buffer size for the retry // retryChannelSize is the buffer size for the retry
@@ -151,12 +152,13 @@ type EngineParams struct {
} }
// Engine processes queued deliveries in the background // Engine processes queued deliveries in the background
// using a bounded worker pool architecture. It owns only // using a bounded worker pool architecture. It also stores each
// the cross-target machinery: the worker pool, the queue // event with its deliveries and creates replay deliveries (see
// and retry channels, the recovery routine, and the persistence // store.go). Beyond that it owns only the cross-target machinery:
// helpers and Scheduler that individual targets rely on. // the worker pool, the delivery and retry channels, the recovery
// Each target type owns its own delivery, including retries, // routine, and the persistence helpers and Scheduler that
// backoff, and circuit breaking. // individual targets rely on. Each target type owns its own
// delivery, including retries, backoff, and circuit breaking.
type Engine struct { type Engine struct {
database *database.Database database *database.Database
dbManager *database.WebhookDBManager dbManager *database.WebhookDBManager
@@ -226,38 +228,6 @@ func New(
return e return e
} }
// 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,
)
}
}
}
// StateAndCooldown implements CircuitBreakers. It returns the state of // StateAndCooldown implements CircuitBreakers. It returns the state of
// the target's circuit breaker and, while the breaker is open, what is // 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 // left of its cooldown; the cooldown is zero once that has passed and
@@ -318,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 // userAgent is the User-Agent header of every http and slack
// delivery request: the program name and the running build's // delivery request: the program name and the running build's
// version. // version.
@@ -915,7 +917,7 @@ func (e *Engine) recoverSingleRetry(
// it stopped working on before then: for example one whose bookkeeping // it stopped working on before then: for example one whose bookkeeping
// write failed, whose attempt a stop cut off, that was still queued // write failed, whose attempt a stop cut off, that was still queued
// when the engine last stopped, whose task was dropped because the // 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. // worker could not open the webhook's database.
// //
// At start every pending delivery is taken, however recent, so one a // At start every pending delivery is taken, however recent, so one a
+3 -3
View File
@@ -492,7 +492,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
) )
task.TargetType = database.TargetTypeLog task.TargetType = database.TargetTypeLog
s.Engine.Notify([]delivery.Task{task}) s.Engine.ExportNotify([]delivery.Task{task})
iWaitForDelivered(t, s.WebhookDB, d.ID) iWaitForDelivered(t, s.WebhookDB, d.ID)
@@ -1452,7 +1452,7 @@ func TestDeliverHTTP_NoURL(t *testing.T) {
iAssertResultFailed(t, db, del.ID) iAssertResultFailed(t, db, del.ID)
} }
// --- Notify batching --- // --- notify batching ---
func TestNotify_MultipleTasks(t *testing.T) { func TestNotify_MultipleTasks(t *testing.T) {
t.Parallel() t.Parallel()
@@ -1467,7 +1467,7 @@ func TestNotify_MultipleTasks(t *testing.T) {
} }
} }
e.Notify(tasks) e.ExportNotify(tasks)
for i := range 5 { for i := range 5 {
select { select {
+6 -6
View File
@@ -29,7 +29,7 @@ const (
// OnStart before the caller may enqueue work. A worker pool // OnStart before the caller may enqueue work. A worker pool
// wrongly rooted in the already-done hook context has nothing // wrongly rooted in the already-done hook context has nothing
// but ctx.Done() ready in its select, so it is deterministically // 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() // race the pool's very first select, in which a ready ctx.Done()
// and a ready deliveryCh are chosen between at random and a // and a ready deliveryCh are chosen between at random and a
// doomed pool still delivers. // doomed pool still delivers.
@@ -182,7 +182,7 @@ func TestEngine_WorkersOutliveStartHookContext(t *testing.T) {
// routine's run at start cannot enqueue it. // routine's run at start cannot enqueue it.
deliveryID, task := seedLogTask(t, s) deliveryID, task := seedLogTask(t, s)
s.Engine.Notify([]delivery.Task{task}) s.Engine.ExportNotify([]delivery.Task{task})
iWaitForDelivered(t, s.WebhookDB, deliveryID) 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 // Let the pool prove it is running before stopping it, so a
// fast OnStop cannot pass by stopping something already dead. // fast OnStop cannot pass by stopping something already dead.
firstID, firstTask := seedLogTask(t, s) firstID, firstTask := seedLogTask(t, s)
s.Engine.Notify([]delivery.Task{firstTask}) s.Engine.ExportNotify([]delivery.Task{firstTask})
iWaitForDelivered(t, s.WebhookDB, firstID) iWaitForDelivered(t, s.WebhookDB, firstID)
var stopErr error var stopErr error
@@ -232,7 +232,7 @@ func TestEngine_StopHookStopsWorkers(t *testing.T) {
// With every worker gone, a freshly notified task must sit // With every worker gone, a freshly notified task must sit
// untouched in the queue rather than being delivered. // untouched in the queue rather than being delivered.
secondID, secondTask := seedLogTask(t, s) secondID, secondTask := seedLogTask(t, s)
s.Engine.Notify([]delivery.Task{secondTask}) s.Engine.ExportNotify([]delivery.Task{secondTask})
time.Sleep(200 * time.Millisecond) time.Sleep(200 * time.Millisecond)
@@ -383,7 +383,7 @@ func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) {
t, s, 1, database.DeliveryStatusPending, 1, t, s, 1, database.DeliveryStatusPending, 1,
) )
s.Engine.Notify([]delivery.Task{task}) s.Engine.ExportNotify([]delivery.Task{task})
stopDuringAttempt(t, lc, arrived) stopDuringAttempt(t, lc, arrived)
iAssertStatus( iAssertStatus(
@@ -448,7 +448,7 @@ func deliverToArchive(t *testing.T, s iSetup) string {
task.TargetID = tgt.ID task.TargetID = tgt.ID
task.TargetType = database.TargetTypeDatabase task.TargetType = database.TargetTypeDatabase
s.Engine.Notify([]delivery.Task{task}) s.Engine.ExportNotify([]delivery.Task{task})
iWaitForDelivered(t, s.WebhookDB, deliveryID) iWaitForDelivered(t, s.WebhookDB, deliveryID)
+2 -2
View File
@@ -218,7 +218,7 @@ func TestNotify_NonBlocking(t *testing.T) {
done := make(chan struct{}) done := make(chan struct{})
go func() { go func() {
e.Notify([]delivery.Task{ e.ExportNotify([]delivery.Task{
{DeliveryID: "overflow-1"}, {DeliveryID: "overflow-1"},
{DeliveryID: "overflow-2"}, {DeliveryID: "overflow-2"},
}) })
@@ -229,7 +229,7 @@ func TestNotify_NonBlocking(t *testing.T) {
case <-done: case <-done:
case <-time.After(2 * time.Second): case <-time.After(2 * time.Second):
t.Fatal( t.Fatal(
"Notify blocked when delivery channel was full", "notify blocked when delivery channel was full",
) )
} }
} }
+5
View File
@@ -158,6 +158,11 @@ func (e *Engine) ExportUpdateDeliveryStatus(
return e.updateDeliveryStatus(webhookDB, d, "", status) return e.updateDeliveryStatus(webhookDB, d, "", status)
} }
// ExportNotify exposes notify.
func (e *Engine) ExportNotify(tasks []Task) {
e.notify(tasks)
}
// ExportProcessNewTask exposes processNewTask. // ExportProcessNewTask exposes processNewTask.
func (e *Engine) ExportProcessNewTask( func (e *Engine) ExportProcessNewTask(
ctx context.Context, task *Task, ctx context.Context, task *Task,
+1 -1
View File
@@ -61,7 +61,7 @@ func (s *inflightSet) retain(deliveryID string) {
// retainIdle takes the first reference on a delivery, and reports // retainIdle takes the first reference on a delivery, and reports
// whether it got it. It fails when the engine already owns the // 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 // recovery routine, say — mutually exclusive rather than merely
// atomic. // atomic.
func (s *inflightSet) retainIdle(deliveryID string) bool { func (s *inflightSet) retainIdle(deliveryID string) bool {
+9 -9
View File
@@ -84,9 +84,9 @@ func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) {
) )
rAgePending(t, s.WebhookDB, d.ID) 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. // no workers are running in this engine.
s.Engine.Notify([]delivery.Task{{ s.Engine.ExportNotify([]delivery.Task{{
DeliveryID: d.ID, DeliveryID: d.ID,
EventID: event.ID, EventID: event.ID,
WebhookID: s.WebhookID, WebhookID: s.WebhookID,
@@ -109,7 +109,7 @@ func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) {
// TestRecoveryAndSweepDoNotDoubleDispatch drives the recovery routine's // TestRecoveryAndSweepDoNotDoubleDispatch drives the recovery routine's
// run at start and a tick of the sweep at once against one aged pending // 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 // either, and ownership is what keeps any two claimants from both
// sending. Before ownership the two runs both dispatched the row. // sending. Before ownership the two runs both dispatched the row.
func TestRecoveryAndSweepDoNotDoubleDispatch(t *testing.T) { func TestRecoveryAndSweepDoNotDoubleDispatch(t *testing.T) {
@@ -224,7 +224,7 @@ func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
body := `{"released":true}` body := `{"released":true}`
s.Engine.Notify([]delivery.Task{{ s.Engine.ExportNotify([]delivery.Task{{
DeliveryID: d.ID, DeliveryID: d.ID,
EventID: event.ID, EventID: event.ID,
WebhookID: s.WebhookID, WebhookID: s.WebhookID,
@@ -248,10 +248,10 @@ func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
} }
// TestNotifyAfterRecoveryDoesNotSendAgain is the startup race of // 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 // 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 // pending, sends it and releases it, and only then does StoreEvent's
// Notify for it arrive. Nothing owns the delivery by then, so Notify // notify for it arrive. Nothing owns the delivery by then, so notify
// takes it. // takes it.
func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) { func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
t.Parallel() t.Parallel()
@@ -289,7 +289,7 @@ func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
body := event.Body body := event.Body
s.Engine.Notify([]delivery.Task{{ s.Engine.ExportNotify([]delivery.Task{{
DeliveryID: d.ID, DeliveryID: d.ID,
EventID: event.ID, EventID: event.ID,
WebhookID: s.WebhookID, WebhookID: s.WebhookID,
@@ -300,7 +300,7 @@ func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
EntrypointID: event.EntrypointID, 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. // run the task.
require.Eventually( require.Eventually(
t, t,
+2 -2
View File
@@ -85,7 +85,7 @@ func (e *Engine) StoreEvent(event *database.Event) (int, error) {
// resubmit counter is what separates the two. // resubmit counter is what separates the two.
e.mtr.EventReceived() e.mtr.EventReceived()
e.Notify(tasks) e.notify(tasks)
return len(tasks), nil return len(tasks), nil
} }
@@ -124,7 +124,7 @@ func (e *Engine) CreateReplay(
return "", err return "", err
} }
e.Notify([]Task{buildTask(dlv, webhookID, &event, target, 1)}) e.notify([]Task{buildTask(dlv, webhookID, &event, target, 1)})
return dlv.ID, nil return dlv.ID, nil
} }
+18 -9
View File
@@ -280,9 +280,10 @@ func TestStoreEvent_LargeBodyIsNotCarried(t *testing.T) {
} }
// TestStoreEvent_RollsBackWhenAWriteFails proves the event, its // TestStoreEvent_RollsBackWhenAWriteFails proves the event, its
// deliveries and its totals are one transaction: when the target totals // deliveries and its totals are one transaction: when the entrypoint
// write fails, after the event and its delivery were written, none of // totals write fails, the last one, after the event, its delivery, the
// them is left and nothing is queued. // target totals and the event totals were written, none of them is left
// and nothing is queued.
func TestStoreEvent_RollsBackWhenAWriteFails(t *testing.T) { func TestStoreEvent_RollsBackWhenAWriteFails(t *testing.T) {
t.Parallel() t.Parallel()
@@ -290,15 +291,19 @@ func TestStoreEvent_RollsBackWhenAWriteFails(t *testing.T) {
storeTestTarget(t, s, "only") storeTestTarget(t, s, "only")
require.NoError(t, require.NoError(t,
s.WebhookDB.Migrator().DropTable(&database.TargetTotals{})) s.WebhookDB.Migrator().DropTable(&database.EntrypointTotals{}))
_, err := s.Engine.StoreEvent(newStoreTestEvent(s, `{"lost":true}`)) _, err := s.Engine.StoreEvent(newStoreTestEvent(s, `{"lost":true}`))
require.Error(t, err) require.Error(t, err)
assert.Zero(t, countRows(t, s.WebhookDB, &database.Event{})) 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.Delivery{}))
assert.Zero(t, storedEventTotals(t, s.WebhookDB).Events) assert.Zero(t, countRows(t, s.WebhookDB, &database.TargetTotals{}))
assert.Zero(t, countRows(t, s.WebhookDB, &database.EntrypointTotals{}))
totals := storedEventTotals(t, s.WebhookDB)
assert.Zero(t, totals.Events)
assert.Nil(t, totals.LastEventAt)
assert.Empty(t, queuedTasks(s.Engine)) assert.Empty(t, queuedTasks(s.Engine))
assert.Zero(t, s.Engine.ExportInflightHeld()) assert.Zero(t, s.Engine.ExportInflightHeld())
} }
@@ -306,13 +311,17 @@ func TestStoreEvent_RollsBackWhenAWriteFails(t *testing.T) {
// TestCreateReplay_CreatesAndQueuesAReplay proves CreateReplay adds a // TestCreateReplay_CreatesAndQueuesAReplay proves CreateReplay adds a
// new pending delivery marked as a replay, counts it in the target's // 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 // totals, queues its task built from the target it was given and the
// stored event, and leaves the delivery it repeats as it was. // 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) { func TestCreateReplay_CreatesAndQueuesAReplay(t *testing.T) {
t.Parallel() t.Parallel()
s := newISetup(t) s := newISetup(t)
target := storeTestTarget(t, s, "replayed") target := storeTestTarget(t, s, "replayed")
event := iSeedEvent(t, s.WebhookDB, s.WebhookID, `{"replay":"me"}`) event := newStoreTestEvent(s, `{"replay":"me"}`)
require.NoError(t, s.WebhookDB.Create(event).Error)
original := iSeedDelivery( original := iSeedDelivery(
t, s.WebhookDB, event.ID, target.ID, t, s.WebhookDB, event.ID, target.ID,
database.DeliveryStatusFailed, database.DeliveryStatusFailed,
@@ -334,7 +343,7 @@ func TestCreateReplay_CreatesAndQueuesAReplay(t *testing.T) {
tasks := queuedTasks(s.Engine) tasks := queuedTasks(s.Engine)
require.Len(t, tasks, 1) require.Len(t, tasks, 1)
assertTaskFor(t, tasks[0], deliveryID, &event, &target) assertTaskFor(t, tasks[0], deliveryID, event, &target)
} }
// TestCreateReplay_RollsBackWhenAWriteFails proves the replay delivery // TestCreateReplay_RollsBackWhenAWriteFails proves the replay delivery
+4 -2
View File
@@ -172,8 +172,10 @@ func loadResubmitSource(
} }
// queueResubmit has the delivery engine store the copy, with a delivery // queueResubmit has the delivery engine store the copy, with a delivery
// for each of the webhook's active targets as they stand now, exactly // for each of the webhook's active targets as they stand now, through
// as it stores a received event. // 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( func (h *Handlers) queueResubmit(
w http.ResponseWriter, w http.ResponseWriter,
r *http.Request, r *http.Request,
+2 -2
View File
@@ -239,8 +239,8 @@ func newTestAppWithConfig(
session.New, session.New,
archive.NewWriters, archive.NewWriters,
// The engine is built on a lifecycle that is never started, // The engine is built on a lifecycle that is never started,
// so no worker runs: every delivery a handler creates stays // so no worker runs: every delivery the engine creates for a
// pending, where the test can read it. // handler stays pending, where the test can read it.
func(p delivery.EngineParams) *delivery.Engine { func(p delivery.EngineParams) *delivery.Engine {
return delivery.New(fxtest.NewLifecycle(t), p) return delivery.New(fxtest.NewLifecycle(t), p)
}, },
+7 -6
View File
@@ -21,8 +21,9 @@ const budget = 64
// batchRunes is how many consecutive code points the charge test logs // 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 // 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 // was too slow for the suite when it ran under the race detector;
// is 271 batches, each logged on two lines, so 542 lines per handler. // 4,096 at a time is 271 batches, each logged on two lines, so 542
// lines per handler.
const batchRunes = 4096 const batchRunes = 4096
// newHandlers are the two handlers internal/logger can install. Time // 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()) batches = append(batches, batch.String())
} }
// What EncodedBytes charges for each batch. Under -race -cover this // What EncodedBytes charges for each batch. When the suite ran
// takes longer than logging the batches, so it is worked out once, // under -race -cover this took longer than logging the batches, so
// by whichever handler finishes logging first, while the other is // it is worked out once, by whichever handler finishes logging
// still logging. // first, while the other is still logging.
charged := sync.OnceValue(func() []int { charged := sync.OnceValue(func() []int {
costs := make([]int, len(batches)) costs := make([]int, len(batches))
+7 -8
View File
@@ -10,14 +10,12 @@ import (
"sneak.berlin/go/webhooker/internal/server" "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 // runs. The window it aims at is the gap between the OnStart hook
// returning and the serving goroutine reaching its first field // returning and the serving goroutine reaching its first field
// access, which is microseconds wide. The race detector reports an // access, which is microseconds wide, so a single cycle can miss it
// unsynchronised pair whenever it observes one, but it has to observe // purely on scheduling. Repetition makes hitting it reliable; the
// one, so a single cycle can miss purely on scheduling. Repetition // collaborators are built once, so the cycles themselves are cheap.
// makes the observation reliable; the collaborators are built once,
// so the cycles themselves are cheap.
const earlyStopIterations = 25 const earlyStopIterations = 25
// TestEarlyShutdown_NoPanicAndNoRace stops the application // TestEarlyShutdown_NoPanicAndNoRace stops the application
@@ -38,8 +36,9 @@ const earlyStopIterations = 25
// //
// httpServer is now built in New, on the constructing goroutine, so // httpServer is now built in New, on the constructing goroutine, so
// it is written before any hook exists and can never be nil. // it is written before any hook exists and can never be nil.
// sentryEnabled is atomic. This test is what catches either one // sentryEnabled is atomic. This test catches the nil dereference
// coming back — under -race, which is how the suite runs. // 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) { func TestEarlyShutdown_NoPanicAndNoRace(t *testing.T) {
t.Parallel() t.Parallel()
-3
View File
@@ -13,9 +13,6 @@ import (
// TestBaseTemplateScriptsAreServed walks every /s/ script the base // TestBaseTemplateScriptsAreServed walks every /s/ script the base
// template loads on each page and fetches it through the real router. // template loads on each page and fetches it through the real router.
// Alpine.js is extracted from its tarball in 3p/ at build time, so the
// file is not in the tree: this is the check that the page still gets
// the JavaScript it asks for.
func TestBaseTemplateScriptsAreServed(t *testing.T) { func TestBaseTemplateScriptsAreServed(t *testing.T) {
t.Parallel() t.Parallel()
-17
View File
@@ -1,17 +0,0 @@
#!/bin/sh
# script/assets: extract Alpine.js from its npm package tarball, committed
# in 3p/, to static/js/alpine.min.js, where go:embed reads it. The package
# is @alpinejs/csp, Alpine's build for pages whose Content-Security-Policy
# forbids eval. The extracted file is not committed. make build, make dev
# and the Dockerfile's lint, test and build stages run this first.
set -eu
ROOT="$(cd "$(dirname "$0")/.." && pwd -P)"
main() {
cd "$ROOT"
tar -xzOf 3p/alpinejs-csp-3.14.9.tgz package/dist/cdn.min.js \
>static/js/alpine.min.js
}
main "$@"
+11
View File
File diff suppressed because one or more lines are too long
+5 -3
View File
@@ -6,9 +6,11 @@ import (
) )
// Static holds the CSS and JavaScript files the web UI's pages load. They // Static holds the CSS and JavaScript files the web UI's pages load. They
// are named one by one so that a missing js/alpine.min.js, which make // are named one by one, so a missing file fails the build instead of
// assets extracts and git does not track, fails the build instead of // leaving the pages without it.
// leaving the pages without Alpine.js. //
// js/alpine.min.js is Alpine.js, committed unchanged. It is
// package/dist/cdn.min.js from the npm package @alpinejs/csp 3.14.9.
// //
//go:embed css/tailwind.css css/style.css js/app.js js/alpine.min.js //go:embed css/tailwind.css css/style.css js/app.js js/alpine.min.js
var Static embed.FS var Static embed.FS