Outbound webhooks

battery/webhook delivers signed POST requests to subscriber URLs with retry-with-backoff and a dead-letter terminal state. It's the external mirror of framework/event (which is internal pub/sub). Use events for in-process listeners, webhooks for talking to other systems.

webhook_deliveries_total and webhook_failures_total each count a delivery dead-lettered after exhausting its claim budget, even when no HTTP request is sent. Each failed attempt increments deliveries before failures. Counters are sampled separately, so a scrape overlapping a failed attempt can transiently report more failures than deliveries.

Wiring

import "github.com/DonaldMurillo/gofastr/battery/webhook"

store := webhook.NewMemoryStore()
mgr := webhook.New(store, webhook.Options{
    // All optional; defaults shown.
    // MaxAttempts:          6,
    // Backoff:              []time.Duration{30s, 1m, 5m, 15m, 1h, 3h},
    // PollInterval:         1 * time.Second,
    // MaxResponseBodyBytes: 64 << 10,   // 64 KiB; protects against malicious receivers
    // AllowPrivateNetworks: false,      // SSRF guard; flip true for dev/tests only
    // Retention:            0,          // terminal-row reaping; see "Retention" below
})
mgr.Start()
defer mgr.Stop(ctx)

Stop(ctx) cancels any in-flight HTTP attempt via the worker's context and waits for the worker goroutine to drain. Pass a tight context if shutdown speed matters more than letting a hung receiver complete. On a LeasedStore, the deliveries of the claimed batch that had not started go back with their attempt refunded, so the next process sends them at once instead of after the lease. So does the POST that Stop cancelled mid-flight: a shutdown is not a failed attempt, and a delivery on its last attempt is retried by the next process rather than dead-lettered. The receiver may have seen that POST, so it can arrive twice; dedupe on X-GoFastr-Delivery-ID.

Register a subscriber and publish:

sub, err := mgr.Subscribe(ctx, webhook.Subscriber{
    URL:    "https://customer.example.com/hooks",
    Secret: "share-this-with-the-customer",
    Events: []string{"orders.**", "shipments.created"},
})
// ...
queued, err := mgr.Publish(ctx, "orders.created", []byte(`{"id":42}`))

Authenticating to the receiver

Many receivers check their own credential before they look at a signature. Set Subscriber.Auth and every delivery sends it beside the signature:

sub, err := mgr.Subscribe(ctx, webhook.Subscriber{
    URL:    "https://customer.example.com/hooks",
    Secret: "share-this-with-the-customer",
    Auth:   webhook.Auth{Kind: webhook.AuthAPIKey, Token: apiKey}, // X-API-Key: <apiKey>
})

Subscribe refuses a malformed credential: an unknown kind, a missing token, a value holding a control byte, or an API key header the delivery sets itself (Host, Content-Type, User-Agent, any X-GoFastr-*). Its errors never repeat the credential.

A credential needs an https:// URL. Subscribe refuses Auth on an http:// URL, and a subscriber that reached the store with one anyway fails each attempt without sending: the token or password would cross the network in clear text. A receiver on loopback or a trusted LAN segment can opt in with Options.AllowPlaintextAuth = true.

A delivery that carries a credential never follows a redirect, even through a caller-supplied HTTPClient that does. Go drops Authorization on a cross-host redirect but not X-API-Key or any other header, so a receiver that answers 3xx would otherwise hand the key to whatever host it names. The 3xx is recorded as a failed attempt.

Bearer and API-key credentials refuse Username and Password, even when only one is set; those fields apply to Basic authentication only. Misconfigured credentials fail at Subscribe, rather than losing fields when the delivery is sent.

The SQLStore seals Auth with the same codec as Secret (see Secret encryption at rest). Subscribe replaces the whole subscriber, so re-subscribing an ID with a zero Auth clears its credential; read the subscriber first when you change only its URL. A Store of your own must persist Auth, or deliveries go out without it.

Publish returns the number of subscribers the event was queued for. The actual HTTP POST happens on the manager's worker goroutine.

A pending delivery with a zero NextAttemptAt is immediately due in both MemoryStore and SQLStore, whether added or updated through Store. SQL NULL schedules follow that same rule, including existing rows. Future schedules remain delayed; dead and successful rows are not due.

When the payload carries one user's data, publish with PublishFor:

queued, err := mgr.PublishFor(ctx, userID, "orders.created", payload)

Each delivery records Owner, and the SQLStore registers a data eraser, so App.EraseUserData deletes that user's deliveries, sent or not. Publish is PublishFor with no owner.

Event names (and the glob patterns Subscribe accepts) are validated: empty names are refused, and so is any character from the unsafe control set — C0, DEL, the C1 block (U+0080–U+009F, whose 8-bit CSI/OSC forms drive terminal escapes and which Go's header writer would otherwise carry raw in X-GoFastr-Event), and the zero-width/bidi invisible set. No control character reaches an outbound header or a persisted row through an event name.

SSRF guard

By default, Subscribe rejects URLs that target internal infrastructure: RFC1918 (10.*, 192.168.*, 172.16-31.*), loopback (127.0.0.0/8, ::1), link-local (169.254.0.0/16, including the AWS/GCP metadata endpoint at 169.254.169.254), the unspecified address (0.0.0.0), localhost, *.localhost, *.internal, and metadata.google.internal. Schemes other than http/https are always rejected.

The guard runs both on the hostname directly and on every resolved IP when the host is a name (DNS lookup at subscribe time). It also runs again at dial time: the default delivery client's Transport.DialContext installs a net.Dialer.Control hook that re-validates the actual resolved IP at connect time. This closes the DNS-rebinding / TOCTOU window where a host validates public at Subscribe and is then re-pointed at 127.0.0.1 / 169.254.169.254 / an RFC1918 address before the worker fires; the connection is refused before any bytes leave the process. The default transport ignores HTTP_PROXY/HTTPS_PROXY: through a proxy the hook would see only the proxy's address. To deliver through an egress proxy, supply your own client (next paragraph).

Supplying your own Options.HTTPClient (proxy, tracing, custom timeout) does not drop the guard: New wraps the client with a per-request check that resolves the delivery target and refuses internal IPs before your transport runs. Your transport itself is used verbatim: a private egress proxy, SSH tunnel, or custom dialer keeps working, since it is the target that must be public, not the route to it. Only AllowPrivateNetworks: true opts out.

For development and tests, opt out via Options.AllowPrivateNetworks = true. This disables both the subscribe-time and the dial-time IP checks. The scheme guard still applies: file://, gopher://, etc. are always refused.

Receivers on the same LAN or machine

A receiver at 192.168.1.20, 10.0.0.5, localhost or a *.internal name is refused unless the manager opts out with Options.AllowPrivateNetworks = true. The setting covers the whole manager, not one subscriber: once it is on, any subscriber URL may name an internal address, the cloud metadata endpoint included. Turn it on only when the people who can register subscribers are trusted with your network. When subscribers come from untrusted tenants, run a separate manager with the guard on for them.

Delivered request shape

Binding the timestamp into the signed material is the same convention Stripe uses; receivers reject captured payloads outside their tolerance window so a leaked delivery cannot replay forever. The event name and the delivery id are signed too, so a captured delivery cannot be re-sent under another event or id inside that window. The parts are joined with a newline (\n), which a header value cannot carry; event names contain dots, so a dot would let one event, id and body read as another.

A receiver written in another language computes the same HMAC:

material = f"{t}\n{event}\n{delivery_id}\n".encode() + body
expected = hmac.new(secret.encode(), material, hashlib.sha256).hexdigest()
ok = hmac.compare_digest(expected, v1) and abs(time.time() - t) <= 300

The body is whatever you passed to Publish; application/json isn't required, but the header advertises it.

No traceparent header is sent. The default delivery client records an OpenTelemetry client span per attempt (HTTP POST, URL without its query string) but does not pass trace context to the subscriber, whose server is not yours; see observability.

Verifying inbound webhooks (receiver side)

body, _ := io.ReadAll(r.Body)
ok := webhook.VerifyTimestamped(
    secret,
    r.Header.Get(webhook.SignatureHeader),
    r.Header.Get(webhook.EventHeader),
    r.Header.Get(webhook.DeliveryIDHeader),
    body,
    5*time.Minute, // tolerance window
)
if !ok {
    http.Error(w, "bad signature", http.StatusUnauthorized)
    return
}

VerifyTimestamped is constant-time and rejects:

• an empty or whitespace-only secret
• header missing t= or v1= fields
• a timestamp outside the tolerance window (replay defense)
• an event or delivery id carrying a newline
• mismatched HMAC (any tampering with the body, event or delivery id)

Response handling

Each attempt reads at most Options.MaxResponseBodyBytes of the receiver's response body (default 64 KiB). A malicious or buggy receiver cannot exhaust the manager's memory by returning a multi-gigabyte body.

The delivery keeps what the receiver said, so a refusal can be diagnosed without a packet capture:

• Delivery.ResponseStatus: the status of the last attempt that got a response; 0 when it got none (a dial error, a timeout).
• Delivery.ResponseBody: the first Options.ResponseBodyLimit bytes of that body (default 1024; negative keeps none). Invalid UTF-8 is replaced and control bytes are percent-encoded, so printing it cannot drive a terminal.

LastError stays short (http 401). A subscriber's own credential is replaced with [redacted] wherever the receiver echoes it back (an error page that prints the Authorization or API key header), since the body column is not sealed like the auth column. The body is stored with the delivery, so a receiver that echoes the payload or a secret back puts it in the store; keep the limit at what an error message needs. The SQLStore keeps both in response_status and response_body, which the data exporter includes and App.EraseUserData deletes with the row.

Event matching

Subscriber Events is a list of glob patterns. Two wildcards are supported:

Pass an empty list to Subscribe and it defaults to ["*"].

Retry semantics

A delivery succeeds when the receiver returns 2xx. Anything else (4xx, 5xx, network error) marks the delivery failed, schedules the next attempt using Options.Backoff, and the worker picks it up at the scheduled time.

After Options.MaxAttempts attempts the delivery transitions to dead. Every claim consumes one attempt: the store bumps the row's counter when ClaimDueDeliveries reserves it, so a worker that crashes between claim and settle (deploy kill, OOM) still burns budget and a crash loop terminates at MaxAttempts instead of redelivering forever. A claimed delivery whose count is already past MaxAttempts (earlier claims spent the budget and their workers died before settling) is marked dead without a request, so MaxAttempts: 1 means at most one POST even across crashes. Settles never rewrite the counter — two overlapping runners (a worker whose lease expired mid-POST, the designed recovery re-claim) each count their own POST, so the recorded attempts always equal the deliveries that ran. A transient error looking the subscriber up (a store blip, a cancelled worker context) is not one of those: it records failed and schedules the next attempt, since dead is reserved for an exhausted attempt budget or a subscriber that is definitively gone or inactive. A row already success or dead is fenced: a stale claimant's late non-terminal settle is a no-op, never a re-queue of a delivered row. A retry or dead settle from a claim the row has moved past is fenced too: once another worker re-claims, the late retry would replace that worker's lease, and a third worker could claim the delivery while the second is still sending it; once an operator resets the delivery, a late give-up would dead-letter the replay before it ran. A success settle is never lease-fenced, since the receiver already has the payload. A settle for a row retention already reaped does nothing, so completed work never becomes deliverable again. Nothing but a success replaces a success row: a claimant past the budget that dead-letters while the worker it outran is still mid-POST does not mark that delivery dead once it lands. Dead deliveries are not retried; they remain in the store for inspection / replay via your own admin tooling.

If a subscriber is removed while a delivery for it is pending, the delivery transitions straight to dead with LastError = "subscriber gone or inactive".

Retention (optional)

Successful delivery rows and processed inbound envelopes are terminal — they can never become deliverable again — and by default they live forever. Every Publish writes one row per matching subscriber and every verified inbound POST writes an envelope (payload up to 1 MiB), so routine traffic grows the tables until the disk fills unless something reaps them.

Set Options.Retention (a time.Duration, zero = off, the outbox-style opt-in) and the manager's worker tick reaps successful deliveries — and, when you also wire Options.InboundStore, processed envelopes — whose updated_at is older than the window. Dead deliveries and failed envelopes are deliberately never reaped: replay and post-mortems need the body, so retention is not a dead-letter TTL. A negative value is clamped to zero (off).

mgr := webhook.New(store, webhook.Options{
    Retention:    30 * 24 * time.Hour,
    InboundStore: inboundStore, // swept by the same pass; optional
})

The worker survives a panicking store: every store call on the delivery loop runs under a recover guard that logs the control-byte-scrubbed panic value through Options.Logger, so a custom Store implementation that panics costs the affected tick or attempt (due rows stay pending and retry, at-least-once) instead of the process. A subscriber row whose events column no longer parses as JSON is surfaced as a lookup error rather than silently matching zero events.

Pausing a subscriber

Set Paused: true in the Subscriber you pass to Subscribe to register it inactive; Publish skips paused subscribers and no deliveries are queued. To resume, re-subscribe (Subscribe upserts on ID) with Paused: false. The default for a new subscription is active.

Stores

webhook.Store covers subscribers and deliveries:

type Store interface {
    AddSubscriber(ctx, Subscriber) error
    GetSubscriber(ctx, id) (*Subscriber, error)
    ListSubscribers(ctx) ([]Subscriber, error)
    DeleteSubscriber(ctx, id) error

    AddDelivery(ctx, Delivery) error
    UpdateDelivery(ctx, Delivery) error
    ListDeliveries(ctx, subscriberID, limit) ([]Delivery, error)
    DueDeliveries(ctx, now, limit) ([]Delivery, error)
}

MemoryStore copies subscriber event slices and delivery payload bytes when storing and returning records, so editing a returned record requires an explicit store write to change delivery state.

Two stores are bundled:

• NewMemoryStore(): in-process maps, suitable for tests and single-instance apps that tolerate restart loss.
• NewSQLStore(db, opts...): SQL-backed (sqlite + postgres), creates webhook_subscribers and webhook_deliveries on first use. Options: - WithSQLSubscribersTable(name) / WithSQLDeliveriesTable(name) : override table names. - WithSQLSecretCodec(codec): encrypt subscriber secrets at rest (see Secret encryption below). - WithSQLAllowPlaintext(): opt into plaintext storage (NoopSecretCodec).

NewSQLStore fails closed: given neither option it returns an error rather than silently storing subscriber secrets in cleartext.

Both bundled stores implement the optional LeasedStore interface:

type LeasedStore interface {
    Store
    ClaimDueDeliveries(ctx, now, limit, leasePeriod) ([]Delivery, error)
}

ClaimDueDeliveries atomically reserves rows for the calling worker, pushes their NextAttemptAt forward by leasePeriod, and consumes one attempt per claimed row, so a concurrent Manager sees them as not-yet-due and skips them. The Manager auto-detects the interface and uses the claim path, which makes multi-instance deployments safe against double delivery. Set Options.LeasePeriod (default 30s) above your worst-case handler latency. A LeasePeriod no longer than HTTPClient.Timeout (10s on the default client) is raised by New to twice that timeout and logged, so a slow receiver is not sent the same delivery by a second Manager. A worker sends a claimed batch one delivery at a time; when less of the lease is left than the HTTP client's timeout (capped at half the lease), it hands that delivery and the rest of the batch back, attempt refunded, for the next tick. A store without LeasedStore still counts each attempt and dead-letters at MaxAttempts.

A leasePeriod of 0 selects the 30 s default; a negative one is rejected by both bundled stores — a negative lease can only be a sign or unit error, and folding it onto the default would hand the caller a longer lease than they asked for.

SQLite stores a timestamp as text with its offset and compares the text, so the SQL stores bind every timestamp in UTC whatever zone the caller's clock carries. Workers in different zones, or a lease that spans a DST change, then read each other's leases and due times correctly. Opening a SQLStore on SQLite rewrites in UTC the due times of pending rows an earlier release stored in the writer's zone. A replica still running that earlier release keeps writing its own zone until it is upgraded.

Secret encryption at rest

The SQL store can encrypt webhook_subscribers.secret, and the auth column holding Subscriber.Auth, so a DB snapshot doesn't hand out HMAC keys or receiver credentials:

key := mustReadFromKMS() // 32 bytes for AES-256
codec, err := webhook.NewAESGCMSecretCodec(key)
if err != nil { /* ... */ }
store, _ := webhook.NewSQLStore(db, webhook.WithSQLSecretCodec(codec))

The encoded format is wbenc:v1:<base64(nonce||ciphertext)>. Rows without the wbenc: prefix are returned as-is on read so an existing deployment can roll the codec without a one-shot rewrite job; each subscriber's secret is re-encrypted the next time the row is upserted.

Key rotation: re-encrypt subscribers by reading them through a codec that decrypts with the old key and encrypts with the new one (typically a small wrapper around two NewAESGCMSecretCodec instances), then write each subscriber back through AddSubscriber. The package keeps a single-key codec primitive on purpose; a key ring belongs in your KMS adapter.

Auto-bridging from framework/event

To send every internal event to webhook subscribers without writing per-event glue, call webhook.Bridge:

cancel := webhook.Bridge(app.Events(), mgr)              // entity.created/updated/deleted
defer cancel()

cancel := webhook.Bridge(app.Events(), mgr, "orders.**") // custom list

The bridge subscribes one handler per event type, marshals event.Event.Data to JSON, and calls Manager.PublishFor with the owner the event names under event.DataKeyOwnerID (CRUD events on an entity with an owner field carry it), so erasing that user deletes the deliveries that copied their row. The owner is read before WithBridgeRedactor runs, so a redactor that drops it from the payload keeps the tag. The returned cancel detaches every subscription at once; call it before Manager.Stop so no Emit lands after the worker exits.

With several replicas on one fanout (framework.WithFanout), the bridge queues an event only on the replica that emitted it and skips the copies other replicas receive (event.IsRemote). Every replica that emits events must run the bridge, and the replicas must share one Store: with a store per replica, an event emitted on one never reaches subscribers recorded only in another's.

Inbound ingestion

Everything above is outbound: you calling other systems. The receiving side is IngestHandler: an HTTP handler that verifies a request, persists an envelope, acks immediately, and hands the real work to the queue battery. It is the inbound mirror of the Manager.

Wiring

import "github.com/DonaldMurillo/gofastr/battery/webhook"
import "github.com/DonaldMurillo/gofastr/battery/queue"

store := webhook.NewMemoryInboundStore()
q := queue.NewMemoryQueue(4)

// Build the ingestion endpoint. The handler does NOT touch the store
// until the signature checks out.
h, err := webhook.IngestHandler(webhook.IngestConfig{
    Source:       "github",
    Verifier:     webhook.TimestampedVerifier(secret, 5*time.Minute),
    Store:        store,
    Queue:        q,
    JobType:      "webhook.inbound",
    MaxBodyBytes: 1 << 20,             // default 1 MiB
    KeepHeaders:  []string{"X-GitHub-Event", "X-GitHub-Delivery"},
    DedupeKeyFunc: func(r *http.Request, _ []byte) string {
        return r.Header.Get("X-GitHub-Delivery")
    },
})
mux.Handle("/hooks/github", h)
q.Start()
defer q.Close()

// Register the queue handler that does the actual work.
q.RegisterHandler("webhook.inbound", webhook.ProcessInbound(store, func(ctx context.Context, e webhook.InboundEnvelope) error {
    // ...your business logic, with the verified payload in hand
    return nil
}))

IngestConfig.Queue is optional. Leave it nil to persist-and-forget (the envelope is still acked 202); set it to fan processing off the request path.

IngestConfig.Owner maps a verified event to the local user it concerns, for example by looking a provider account id up in the app's own table. The envelope stores the result in InboundEnvelope.Owner, and SQLInboundStore registers a data eraser, so App.EraseUserData deletes that user's payloads. Nil, or an empty return, names nobody. The handler runs only after the signature checks out. The 202 ack sets Cache-Control: no-store.

Request flow

1. Non-POST → 405.
2. Body read through http.MaxBytesReader; oversize → 413.
3. Verify, then persist: an unverified payload is never written to the store. A verification failure responds 401 with a generic body (signature verification failed); no header value or reason detail is echoed.
4. Dedupe: if DedupeKeyFunc returns a non-empty key the store has already seen for this source, the handler acks 200 immediately without re-persisting or enqueuing; idempotent redelivery.
5. Persist the envelope as received, allowlisted headers only. With a queue wired, the dedupe key is deliberately not stored yet.
6. Enqueue a queue.Job{Type: JobType, Payload: {"envelope_id", "source"}}.
7. Register the dedupe key on the envelope; only now that the event is durably queued may redeliveries be dedupe-acked. A worker may already hold the envelope, so the bundled stores write the key alone (RegisterDedupeKey), at any status, and a settle that carries no key keeps the stored one. A host store without RegisterDedupeKey gets the key through UpdateEnvelope.
8. Respond 202 with {"id": "<envelope id>"}.

On enqueue failure the handler responds 500 (the sender retries) and best-effort marks the just-written envelope failed with LastError as a forensic record. Because the key is registered only after a successful enqueue (step 7), an envelope that never reached the queue can never dedupe-ack the sender's retry; the redelivery persists and enqueues a fresh copy even if the forensic marking itself failed. Without a queue, persistence alone is durable acceptance, so the key is stored with the envelope up front. Failures of these best-effort updates are reported through IngestConfig.Logger (default log.Printf).

Verifiers

Two are bundled:

• TimestampedVerifier(secret, tolerance): wraps VerifyTimestamped over X-GoFastr-Signature, reading the event and delivery id from their headers. The timestamp, event and id are bound into the signed material, so a captured request can't replay past tolerance or be re-sent under another event or id. Preferred whenever the sender supports it.
• HMACSHA256Verifier(header, prefix, secret): GitHub-style <prefix><hex-hmac-sha256-of-body> (e.g. header X-Hub-Signature-256, prefix sha256=). Uses hmac.Equal, but note there is no timestamp binding; it offers no replay defense. Use it for providers that don't send a timestamp; pair it with a short dedupe window if you can.

Both verifiers reject every request when secret is empty. An HMAC keyed with "" is a signature anyone can compute, so a receiver wired from an unset environment variable answers 401 instead of accepting forgeries.

Provide your own InboundVerifier for other schemes (RSA signatures, mTLS-extracted identity, a multi-key rotation lookup). Implementations must be constant-time on the secret.

Envelope lifecycle

ProcessInbound adapts your business function into a queue.Handler that drives the envelope through its states:

received → processing → processed
                   └→ failed  (fn returned an error)

It loads the envelope by envelope_id from the job payload, marks it processing — consuming one attempt store-side (attempts = attempts + 1 when the store implements the claim; both bundled stores do), so two overlapping runs under the queue's lease expiry each count — before calling your function, then processed on success or failed (with LastError set) on error. It returns your function's error so the queue's own retry/backoff can reschedule it. failed is not final: the retry marks the envelope processing again, consumes an attempt, and records its own error. Only processed is fenced, so a stale runner's late failure cannot overwrite a success.

A job for an envelope already processed (redelivered after a crash before the queue's ack, or after a lease expiry) is acked without calling your function. The store-side claim reports whether it marked the envelope, so one that settled or was deleted between the load and the claim is skipped too. Two runs that overlap while the envelope is still processing both call your function, so keep it idempotent.

Stores

Two InboundStore implementations ship:

• NewMemoryInboundStore(): in-process map; tests and single-instance.
• NewSQLInboundStore(db, opts...): SQL-backed (sqlite + postgres), creates webhook_inbound on first use. WithInboundTable(name) overrides the table name. Headers persist as a JSON TEXT column.

Dedupe is enforced application-side via SeenDedupeKey rather than a database unique constraint: a portable partial index (Postgres) can't be written in one DDL that also works on SQLite, and a plain unique index would forbid two legitimately-undedupe'd (empty-key) requests from coexisting. The (source, dedupe_key) index keeps the lookup cheap, but there is an inherent check-then-insert race window under concurrency; make your ProcessInbound handler idempotent to absorb the rare duplicate.

Common mistakes

• Don't pass the same secret to every subscriber. Generate one per subscriber on registration so a leaked secret only exposes one endpoint.
• Don't trust an unsigned request. Even if the URL is private, rotate keys via the same code path you use for public consumers; signature checks are cheap insurance.
• Don't put a database call on the publish path. Publish writes one delivery row per matching subscriber. If you have many subscribers per event, accept the latency or denormalize.
• Don't catch up old events at startup by replaying everything. Use the DueDeliveries query for genuinely retryable rows; events that pre-date the subscriber should be backfilled deliberately, not resurrected by the retry loop.
• Multi-instance writers need the lease path. The bundled SQL store implements LeasedStore: on Postgres via FOR UPDATE SKIP LOCKED inside an UPDATE … RETURNING, on SQLite via a serializable BEGIN IMMEDIATE transaction. The Manager automatically uses it when the store implements the interface. Custom stores that don't implement LeasedStore are safe for single-instance deployments only; concurrent workers against a plain DueDeliveries-only store can double-deliver.
• The bridge calls Publish synchronously inside the emitter's goroutine. With a SQL store, that means each Emit does a write per matching subscriber before returning. If the emitter is a hot path, switch to EmitAsync or fork a goroutine in your own subscribe handler.