From 599aac99efe5677aab2096b4e74c215344851813 Mon Sep 17 00:00:00 2001 From: GatisB Date: Sun, 6 Sep 2026 19:52:24 +0300 Subject: [PATCH 1/2] Drop the in-tree DCO workflow: the sign-off is enforced by the DCO app and the branch ruleset The workflow shipped with the initial code as the only check. The repository's develop branch now carries a ruleset requiring the DCO app's check, so the in-tree copy was a duplicate advisory run; CONTRIBUTING.md still asks for the sign-off, and the ruleset is what refuses a pull request without it. Signed-off-by: GatisB --- .github/workflows/dco.yml | 54 --------------------------------------- 1 file changed, 54 deletions(-) delete mode 100644 .github/workflows/dco.yml diff --git a/.github/workflows/dco.yml b/.github/workflows/dco.yml deleted file mode 100644 index e9b4094..0000000 --- a/.github/workflows/dco.yml +++ /dev/null @@ -1,54 +0,0 @@ -# Every commit in a pull request must carry a Signed-off-by line matching its own -# author — the Developer Certificate of Origin. CONTRIBUTING.md tells contributors -# this check exists and that a pull request without it fails; this is that check. -# -# It reads the commits with git rather than calling a third-party action, so the -# gate adds no dependency to audit or pin. - -name: dco - -on: - pull_request: - -permissions: - contents: read - -jobs: - dco: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 - with: - fetch-depth: 0 - - - name: Every commit carries a sign-off matching its author - shell: bash - env: - BASE: ${{ github.event.pull_request.base.sha }} - run: | - missing=0 - while read -r sha; do - # A merge commit is generated by git, not written by an author, so it - # carries no trailer and none is expected. Two parents means merge. - if [ "$(git rev-list --parents -n 1 "$sha" | wc -w)" -gt 2 ]; then - continue - fi - # No pipe into grep here, deliberately: this step runs under `pipefail`, - # where `git show | grep -q` returns git's SIGPIPE status (141) once grep - # exits early — which would report a correctly signed commit as missing. - author="$(git show -s --format='%an <%ae>' "$sha")" - signoffs="$(git show -s --format='%(trailers:key=Signed-off-by,valueonly)' "$sha")" - if [[ "${signoffs,,}" != *"${author,,}"* ]]; then - echo "::error::$(git show -s --format='%h %s' "$sha") — missing: Signed-off-by: $author" - missing=$((missing + 1)) - fi - done < <(git rev-list "$BASE"..HEAD) - - if [ "$missing" -gt 0 ]; then - echo - echo "$missing commit(s) carry no Signed-off-by line matching their author." - echo "Add it to new commits with: git commit -s" - echo "Add it to commits already made with: git rebase --signoff $BASE" - exit 1 - fi - echo "Every commit carries a Signed-off-by line matching its author." From cfcfe5fa7e64b4b0d90d4c804a5d29375703fabe Mon Sep 17 00:00:00 2001 From: GatisB Date: Tue, 8 Sep 2026 09:46:22 +0300 Subject: [PATCH 2/2] Carry the causing act's correlation id: Event.CorrelationID is stored with the event and sent as X-Correlation-ID on every delivery attempt The library builds on go-platform-kit's propagation package for the header's name and the context helpers, like the platform's other libraries. An event with no correlation id is sent without the header. sql/V2 adds the column a relational Store needs. The Headers.Delivery doc comment now says what the wire does: one id per event per endpoint, the same on every attempt. Version 0.2.0. Signed-off-by: GatisB --- CHANGELOG.md | 21 ++++++++++++ README.md | 22 +++++++++---- dispatcher.go | 11 +++++++ dispatcher_test.go | 41 +++++++++++++++++++++++- go.mod | 2 ++ go.sum | 12 +++++++ sql/README.md | 9 ++++-- sql/V2__webhook_event_correlation_id.sql | 11 +++++++ store_test.go | 7 +++- webhook.go | 15 +++++++-- 10 files changed, 136 insertions(+), 15 deletions(-) create mode 100644 go.sum create mode 100644 sql/V2__webhook_event_correlation_id.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b5d4e9..6177ee4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,27 @@ Notable changes to this library, newest first, per release. Written for whoever hosts the library or receives its deliveries. +## v0.2.0 + +### Added — the causing act's correlation id travels with the delivery + +`Event.CorrelationID` is the correlation id of the request that caused the event, when the host knew +one. It is stored with the event (`MemoryStore` keeps it; a relational `Store` gets the column from +`sql/V2__webhook_event_correlation_id.sql`) and the worker sends it as the platform's `X-Correlation-ID` +header on **every attempt** of every delivery of the event — a retry continues the same thread. An event +with no correlation id (background work) is sent without the header, never with an empty one. The header +name comes from `go-platform-kit`'s `propagation` package, which is now a dependency: the library carries +the platform kit like the platform's other libraries rather than re-declaring a concern the kit owns. + +For receivers: a delivery may now carry `X-Correlation-ID`; quote it when asking the host about a +delivery. Nothing else on the wire changes. + +### Fixed — the delivery header's documentation + +`Headers.Delivery` was documented as "unique per attempt". The worker has always sent the delivery's own +id — one per event per endpoint, the same value on every attempt — and the README said so; the doc +comment now says the same. + ## v0.1.0 Initial code. diff --git a/README.md b/README.md index 1d255da..793c92e 100644 --- a/README.md +++ b/README.md @@ -9,8 +9,11 @@ and a bounded give-up so a dead endpoint never keeps a queue alive forever. go get github.com/gmb-lib/go-webhook ``` -Standard library only. The library fixes everything a **receiver** can observe and leaves to the -**host** what only the host knows: the event body, where things are stored, when the worker runs. +The library fixes everything a **receiver** can observe and leaves to the **host** what only the host +knows: the event body, where things are stored, when the worker runs. It builds on +[go-platform-kit](https://github.com/gmb-lib/go-platform-kit) for the one cross-cutting concern it +shares with every other service and library of the platform — the correlation id and its header name — +and on the standard library for everything else. Because the receiver-facing contract lives here and nowhere else, a host can later move delivery to another process — same package, another host — and no receiver sees a different call. @@ -23,11 +26,13 @@ Every delivery is one HTTP `POST` to the registered endpoint URL: | `Content-Type` | `application/json` | | `Webhook-Signature` | `t=,v1=` — two `v1` values while a secret rotation is in progress | | `Webhook-Event` | the event type, so you can route before parsing the body | -| `Webhook-Delivery` | the delivery id — the same for every attempt of one delivery | +| `Webhook-Delivery` | the delivery id — one per event per endpoint, the same for every attempt of one delivery | +| `X-Correlation-ID` | the correlation id of the act that caused the event (a request a person or another system made), when the host knew one — the same on every attempt; quote it when you ask the host about a delivery. Absent for an event no request caused | | `User-Agent` | the host's name, or `go-webhook/` | The body is the host's event, byte for byte; the event id inside it is stable across attempts, so **de-duplicate on it**. A host may rename the three `Webhook-*` headers; its documentation then says so. +`X-Correlation-ID` is the platform's header and is never renamed. **The signature** is the lowercase hex HMAC-SHA256, keyed with a secret the host gave you when you registered, over the bytes ` "." `. `t` is bound into the signed bytes so a replayed @@ -90,8 +95,10 @@ _ = disp.Subscribe(ctx, webhook.Subscription{ EventTypes: []string{"order.completed"}, // empty = every type }) -// publishing — never waits on a receiver; it only writes deliveries -_, _ = disp.Publish(ctx, webhook.Event{ID: eventID, ClientID: "acme", Type: "order.completed", Payload: body}) +// publishing — never waits on a receiver; it only writes deliveries. CorrelationID is the +// causing request's (propagation.CorrelationID(ctx) in a handler); leave it empty for +// background work and no header is sent. +_, _ = disp.Publish(ctx, webhook.Event{ID: eventID, ClientID: "acme", Type: "order.completed", Payload: body, CorrelationID: correlationID}) // delivering — one goroutine per process, or several processes against a claiming Store go worker.Run(ctx, 5*time.Second, 100, errs) @@ -109,8 +116,9 @@ that a subscription disabled after an event was queued is not sent to. ### Storing it `Store` is five kinds of read and write over subscriptions, events and deliveries; `MemoryStore` is the -complete reference implementation. For a database, [`sql/V1__webhook_tables.sql`](sql/V1__webhook_tables.sql) -is the table shape — copy it into your migrations and map it, column for column, onto the Go types. +complete reference implementation. For a database, [`sql/`](sql/) is the table shape — copy its +migrations into yours and map them, column for column, onto the Go types (`V1` the three tables, `V2` the +event's `correlation_id`). Secrets are stored as **references** into your secret store, never as values; your `Store` resolves them when the worker asks. With several workers, claim rows as you read them (`FOR UPDATE SKIP LOCKED`) so no delivery is sent twice. diff --git a/dispatcher.go b/dispatcher.go index fe4f03f..07a1699 100644 --- a/dispatcher.go +++ b/dispatcher.go @@ -8,6 +8,8 @@ import ( "io" "net/http" "time" + + "github.com/gmb-lib/go-platform-kit/propagation" ) // Dispatcher is the seam a host publishes through. Publish records an event and @@ -254,6 +256,10 @@ func (w *Worker) send(ctx context.Context, sub Subscription, ev Event, d Deliver body := []byte(ev.Payload) ctx, cancel := context.WithTimeout(ctx, w.timeout()) defer cancel() + // The causing act's correlation id rides on the context as well as on the header, so + // a host-supplied Client whose transport reads the platform's context sees the same + // thread (an empty id leaves the context as it is). + ctx = propagation.WithCorrelationID(ctx, ev.CorrelationID) req, err := http.NewRequestWithContext(ctx, http.MethodPost, sub.EndpointURL, bytes.NewReader(body)) if err != nil { return 0, "build request: " + err.Error() @@ -264,6 +270,11 @@ func (w *Worker) send(ctx context.Context, sub Subscription, ev Event, d Deliver req.Header.Set(h.Signature, SignatureHeader(now, body, secrets...)) req.Header.Set(h.Event, ev.Type) req.Header.Set(h.Delivery, d.ID) + // Every attempt of a delivery carries the same correlation id: a retry continues the + // thread the causing act started. An event that no request caused carries none. + if ev.CorrelationID != "" { + req.Header.Set(propagation.HeaderCorrelationID, ev.CorrelationID) + } resp, err := w.client().Do(req) if err != nil { diff --git a/dispatcher_test.go b/dispatcher_test.go index f06a67c..21e3021 100644 --- a/dispatcher_test.go +++ b/dispatcher_test.go @@ -9,6 +9,8 @@ import ( "sync/atomic" "testing" "time" + + "github.com/gmb-lib/go-platform-kit/propagation" ) // fakeClock is a settable time source shared by the dispatcher and the worker. @@ -118,7 +120,8 @@ func TestWorkerRetriesThenDeliversWithVerifiableSignature(t *testing.T) { t.Fatal(err) } payload := []byte(`{"eventId":"e1","type":"signing-request.completed","sequence":3}`) - ds, err := disp.Publish(ctx, Event{ID: "e1", ClientID: "acme", Type: "signing-request.completed", Payload: payload}) + const correlation = "01K5VB0E-the-causing-request" + ds, err := disp.Publish(ctx, Event{ID: "e1", ClientID: "acme", Type: "signing-request.completed", Payload: payload, CorrelationID: correlation}) if err != nil || len(ds) != 1 { t.Fatalf("publish: %v %+v", err, ds) } @@ -193,6 +196,42 @@ func TestWorkerRetriesThenDeliversWithVerifiableSignature(t *testing.T) { if err := Verify(sig, last.body, [][]byte{testSecret("stranger")}, sentAt, 5*time.Minute); err == nil { t.Fatal("a stranger's secret must not verify") } + // The correlation id of the act that caused the event travels on EVERY attempt, the + // same value each time — a retry is the same thread, not a new one. + for i, call := range rcv.calls { + if got := call.headers.Get(propagation.HeaderCorrelationID); got != correlation { + t.Fatalf("attempt %d: %s = %q, want %q", i+1, propagation.HeaderCorrelationID, got, correlation) + } + } +} + +// An event that no request caused — background work — has no correlation id, and the +// delivery then carries no header at all rather than an empty one. +func TestWorkerSendsNoCorrelationHeaderWhenTheEventHasNone(t *testing.T) { + ctx := context.Background() + rcv := &receiver{} + srv := httptest.NewServer(rcv.handler(t)) + defer srv.Close() + store := NewMemoryStore() + clock := &fakeClock{t: time.Unix(1757170123, 0)} + disp := &InProcess{Store: store, Clock: clock.Now} + _ = disp.Subscribe(ctx, Subscription{ID: "s", ClientID: "c", EndpointURL: srv.URL, Enabled: true, Secrets: []Secret{{Value: testSecret("k")}}}) + ds, _ := disp.Publish(ctx, Event{ID: "e", ClientID: "c", Type: "t", Payload: []byte(`{}`)}) + w := &Worker{Store: store, Clock: clock.Now, Jitter: NoJitter} + if n, err := w.RunOnce(ctx, 0); err != nil || n != 1 { + t.Fatalf("run: n=%d err=%v", n, err) + } + if d := mustDelivery(t, store, ds[0].ID); d.Status != StatusDelivered { + t.Fatalf("delivered expected: %+v", d) + } + rcv.mu.Lock() + defer rcv.mu.Unlock() + if len(rcv.calls) != 1 { + t.Fatalf("receiver saw %d calls", len(rcv.calls)) + } + if vals := rcv.calls[0].headers.Values(propagation.HeaderCorrelationID); len(vals) != 0 { + t.Fatalf("no correlation id on the event, yet the header was sent: %q", vals) + } } func TestWorkerDropsOnClientErrorAndDeadLettersWhenExhausted(t *testing.T) { diff --git a/go.mod b/go.mod index d3fe640..bb0823b 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,5 @@ module github.com/gmb-lib/go-webhook go 1.26.6 + +require github.com/gmb-lib/go-platform-kit v1.11.1 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..a532dfe --- /dev/null +++ b/go.sum @@ -0,0 +1,12 @@ +github.com/gmb-lib/go-platform-kit v1.11.1 h1:UG2VTTLByAyH7X1q6cP/RECurFS6wGs9azPxIiUzZtY= +github.com/gmb-lib/go-platform-kit v1.11.1/go.mod h1:/ZCrZUjDBF+CniHa5xah3IfG0P7o6TqMYXK5ojvNkwU= +github.com/go-quicktest/qt v1.102.0 h1:HSQxCeh5YZH3EL3W39ixjtyaEhcWSXQHtHnMBzSs474= +github.com/go-quicktest/qt v1.102.0/go.mod h1:p4lGIVX+8Wa6ZPNDvqcxq36XpUDLh42FLetFU7odllI= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/rogpeppe/go-internal v1.16.0 h1:O9DK+vNMDVGLr2BeZqmpLeMjiMNkuXfcqntWbZV6S5g= +github.com/rogpeppe/go-internal v1.16.0/go.mod h1:DrUVZyrJU+txYW5/1kwtXQSMFio52ZOxX7yM1VHvnxs= diff --git a/sql/README.md b/sql/README.md index 6d2a3ad..3288e4f 100644 --- a/sql/README.md +++ b/sql/README.md @@ -1,9 +1,12 @@ # Table shape for a relational Store `V1__webhook_tables.sql` is the reference DDL behind the library's `Store` interface: three tables -(`subscription`, `event`, `delivery`) whose columns match the Go types one for one. It is not applied -by the library — copy it into your own migration set, put it in your own schema, and implement `Store` -over it in whatever way your service talks to its database. +(`subscription`, `event`, `delivery`) whose columns match the Go types one for one. +`V2__webhook_event_correlation_id.sql` adds the event's `correlation_id` (the causing request's id, +sent as `X-Correlation-ID` on every attempt; nullable). Neither is applied by the library — copy them +into your own migration set in order, put them in your own schema, and implement `Store` over them in +whatever way your service talks to its database. A host that copied `V1` before `V2` existed adds `V2` +as its own next migration. Two things the shape decides on purpose: diff --git a/sql/V2__webhook_event_correlation_id.sql b/sql/V2__webhook_event_correlation_id.sql new file mode 100644 index 0000000..8f12a12 --- /dev/null +++ b/sql/V2__webhook_event_correlation_id.sql @@ -0,0 +1,11 @@ +-- V2: the event remembers the correlation id of the act that caused it. +-- +-- A delivery carries the X-Correlation-ID header of the request that caused the event +-- (a person's or another system's act), the same value on every attempt, so a receiver +-- can quote one id and the host can find the whole thread across its services. For that +-- the id has to survive with the event, not with the process: a worker that restarts +-- between two attempts reads it back from here. NULL when no request caused the event +-- (background work); the header is then not sent. +-- +-- Nullable, no default — a metadata-only add, no table rewrite. +ALTER TABLE webhook.event ADD COLUMN IF NOT EXISTS correlation_id text; diff --git a/store_test.go b/store_test.go index 105abbc..38bcbdc 100644 --- a/store_test.go +++ b/store_test.go @@ -75,11 +75,16 @@ func TestMemoryStoreSubscriptionsEventsAndNotFound(t *testing.T) { t.Fatalf("acme's subscriptions, sorted: %+v", subs) } - _ = m.SaveEvent(ctx, Event{ID: "e1", ClientID: "acme", Type: "x", Payload: []byte(`{"a":1}`)}) + _ = m.SaveEvent(ctx, Event{ID: "e1", ClientID: "acme", Type: "x", Payload: []byte(`{"a":1}`), CorrelationID: "corr-1"}) ev, err := m.Event(ctx, "e1") if err != nil || string(ev.Payload) != `{"a":1}` { t.Fatalf("event round-trip: %v %s", err, ev.Payload) } + // The correlation id is part of the stored event: a store that dropped it would send + // the first attempt with the header and a retry without. + if ev.CorrelationID != "corr-1" { + t.Fatalf("correlation id not stored with the event: %+v", ev) + } _ = m.Enqueue(ctx, []Delivery{{ID: "d2", EventID: "e1"}, {ID: "d1", EventID: "e1"}, {ID: "d9", EventID: "e9"}}) byEvent, _ := m.ByEvent(ctx, "e1") diff --git a/webhook.go b/webhook.go index 36f6b8e..d135c58 100644 --- a/webhook.go +++ b/webhook.go @@ -30,7 +30,7 @@ import ( ) // Version is the library version, sent as the User-Agent when a host does not set one. -const Version = "0.1.0" +const Version = "0.2.0" // Event is one thing that happened. ID must be stable across delivery attempts: a // receiver de-duplicates on it. Payload is the exact body sent — this package neither @@ -44,6 +44,14 @@ type Event struct { // another's events. ClientID string Payload json.RawMessage + // CorrelationID is the correlation id of the act that caused the event — the request + // a person or another system made — when that act had one. It is stored with the + // event and sent as the X-Correlation-ID header (the platform kit's + // propagation.HeaderCorrelationID — the one home of that header's name) on every + // attempt of every delivery of the event, so a receiver can quote it and + // the host can find the whole thread across its services. Empty for an event that + // no request caused (background work): the header is then not sent. + CorrelationID string } // Subscription is one registered endpoint. EventTypes empty means every type. @@ -106,8 +114,9 @@ type Headers struct { Signature string // Event carries the event type, so a receiver can route before parsing the body. Event string - // Delivery carries the delivery attempt id — unique per attempt, while the event id - // inside the body is the same across attempts. + // Delivery carries the delivery id — one per event per endpoint, the same value on + // every attempt of that delivery. The event id inside the body is the same across + // endpoints too; a receiver may de-duplicate on either. Delivery string }