Compare commits
1
Commits
dae20faa24
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
766f33a956 |
+1
-5
@@ -274,12 +274,8 @@ EXPOSE 8080
|
||||
# port.
|
||||
ENV BIND_ADDRESS=0.0.0.0
|
||||
|
||||
# The probe requests the container's own address, not loopback, so a
|
||||
# listener that answers only on loopback (an empty or loopback
|
||||
# BIND_ADDRESS) fails it. hostname -i prints one IPv4 address per
|
||||
# network the container is on; the first is enough.
|
||||
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
|
||||
CMD wget --no-verbose --tries=1 --spider "http://$(hostname -i | cut -d ' ' -f 1):8080/.well-known/healthcheck" || exit 1
|
||||
CMD wget --no-verbose --tries=1 --spider http://localhost:8080/.well-known/healthcheck || exit 1
|
||||
|
||||
ENTRYPOINT ["/usr/local/bin/docker-entrypoint.sh"]
|
||||
CMD ["/app/webhooker"]
|
||||
|
||||
@@ -316,9 +316,10 @@ The value must be an IP address literal:
|
||||
|
||||
- `127.0.0.1` — loopback only (the binary's default). Use this with a reverse
|
||||
proxy on the same host.
|
||||
- `0.0.0.0` — every IPv4 address and no IPv6 address. The image's default; on a
|
||||
bare host, only behind a firewall on the port.
|
||||
- `::` — every IPv6 address and every IPv4 address.
|
||||
- `0.0.0.0` — every IPv4 address. The image's default; on a bare host, only
|
||||
behind a firewall on the port.
|
||||
- `::` — every address, IPv6 and (on Linux, with the default
|
||||
`net.ipv6.bindv6only=0`) IPv4 as well.
|
||||
- A specific address such as `10.0.0.5` — that interface only.
|
||||
|
||||
An **empty** value is treated as unset, as everywhere else here, and takes the
|
||||
@@ -339,8 +340,7 @@ machine that has no such interface — is a valid literal, so it reaches the
|
||||
listener and fails there. The process logs the bind error and exits non-zero
|
||||
rather than staying up with nothing listening. The effective value is in the
|
||||
`bindAddress` field of the startup log line, which is the way to check what a
|
||||
running deployment actually bound. On a bare host, `ss -ltn` lists the socket
|
||||
itself as `0.0.0.0:8080` for a `0.0.0.0` bind and as `*:8080` for `::`.
|
||||
running deployment actually bound.
|
||||
|
||||
#### Metrics credentials
|
||||
|
||||
@@ -668,8 +668,8 @@ Either set a literal or leave the variable out entirely.
|
||||
Overriding `BIND_ADDRESS` to a loopback address in a container — by that route
|
||||
or deliberately — makes the container unreachable from outside its namespace
|
||||
even with `-p`. The published port answers nothing, and the health check fails
|
||||
too: it requests the container's own address (the first one `hostname -i`
|
||||
prints), where a loopback bind is not listening. The container then goes
|
||||
too: it requests `http://localhost:8080`, `localhost` resolves to `::1` first,
|
||||
and a `127.0.0.1` bind is not listening there. The container then goes
|
||||
`unhealthy` about **65 seconds** after start — from
|
||||
`HEALTHCHECK --start-period=5s --interval=30s --retries=3`, so failing probes at
|
||||
5s, 35s and 65s, and `unhealthy` on the third. (Docker's probe cadence during
|
||||
@@ -726,8 +726,8 @@ the credentials.
|
||||
address (directly, or through anything that can rewrite source addresses,
|
||||
such as NAT or a published container port). The `remoteIP` field of the
|
||||
`http request` log line for a request that came through the proxy shows
|
||||
that address; the health check's own lines show the container's own
|
||||
address. See [Trusted proxies](#trusted-proxies).
|
||||
that address; the health check's own lines show `::1`. See
|
||||
[Trusted proxies](#trusted-proxies).
|
||||
- Leave `BIND_ADDRESS` and `DATA_DIR` unset: the image sets `BIND_ADDRESS`
|
||||
to `0.0.0.0`, and `DATA_DIR` defaults to `/var/lib/webhooker`.
|
||||
- Everything else is optional; see [Configuration](#configuration).
|
||||
@@ -1317,22 +1317,20 @@ webhooker solves this by acting as a durable intermediary:
|
||||
nothing is lost if downstream targets are unavailable.
|
||||
|
||||
2. **Guaranteed delivery** — Events are queued for delivery to each configured
|
||||
target. Failed deliveries are retried with configurable backoff. Delivery
|
||||
attempts are logged with status codes, response bodies, and timing.
|
||||
target. Failed deliveries are retried with configurable backoff. Every
|
||||
delivery attempt is logged with status codes, response bodies, and timing.
|
||||
|
||||
**That guarantee is at-least-once, not exactly-once.** When a send reaches
|
||||
its target but the write recording that outcome fails, the delivery is
|
||||
deliberately left in a recoverable state rather than marked done — losing a
|
||||
delivery is the worse failure — so the pending sweep picks it up about
|
||||
fifteen minutes later, or the next restart does, and the target receives a
|
||||
payload it already got. A send still in flight when the service stops is
|
||||
left the same way: its attempt is not recorded, and the next start sends it
|
||||
again (see Shutdown). webhooker adds no delivery identifier of its own to an
|
||||
outbound request, so **make your receiver idempotent** against whatever the
|
||||
payload itself carries.
|
||||
payload it already got. webhooker adds no delivery identifier of its own to
|
||||
an outbound request, so **make your receiver idempotent** against whatever
|
||||
the payload itself carries.
|
||||
|
||||
3. **Observability** — Full request/response logging for every webhook received
|
||||
and for delivery attempts. Prometheus metrics expose volume, latency, and
|
||||
and every delivery attempted. Prometheus metrics expose volume, latency, and
|
||||
error rates. The web UI provides real-time visibility into event flow.
|
||||
|
||||
4. **Fan-out** — A single incoming webhook can be delivered to multiple targets
|
||||
@@ -1770,7 +1768,7 @@ is what bounds it.
|
||||
#### DeliveryResult
|
||||
|
||||
The result of a single delivery attempt. Every attempt (including retries) is
|
||||
individually logged, except one cut off by a stop (see Shutdown).
|
||||
individually logged for full observability.
|
||||
|
||||
| Field | Type | Description |
|
||||
| --------------- | ------- | ----------------------------------------- |
|
||||
@@ -2208,16 +2206,12 @@ This means:
|
||||
buffer rather than spawning more goroutines.
|
||||
- **Independent results** — each worker records its own delivery result in the
|
||||
per-webhook database without coordination.
|
||||
- **Graceful shutdown** — the stop hook cancels the workers' context. A worker
|
||||
may still take tasks already queued before it exits. An `http` or `slack`
|
||||
request in flight is cut off, and one started after the cancel fails at once.
|
||||
Neither is recorded or counted in the metrics, and the next start sends the
|
||||
delivery again (see Shutdown). A `database` or `log` task runs as normal. The
|
||||
stop hook waits for the pool via `lifecycle.WaitForShutdown`, which bounds
|
||||
that wait by fx's stop timeout rather than blocking forever on a wedged
|
||||
worker. On timeout it logs at `ERROR` and returns an error, and the goroutines
|
||||
that did not finish are still running — an unclean shutdown is reported rather
|
||||
than hidden.
|
||||
- **Graceful shutdown** — cancel the context, workers finish their current task
|
||||
and exit. The stop hook waits for the pool via `lifecycle.WaitForShutdown`,
|
||||
which bounds that wait by fx's stop timeout rather than blocking forever on a
|
||||
wedged worker. On timeout it logs at `ERROR` and returns an error, and the
|
||||
goroutines that did not finish are still running — an unclean shutdown is
|
||||
reported rather than hidden.
|
||||
|
||||
**Recovery paths:**
|
||||
|
||||
@@ -2323,7 +2317,7 @@ getting anywhere.
|
||||
| Metric | Type | Meaning |
|
||||
| -------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `webhooker_events_received_total` | counter | Events received and durably stored. Compare against the delivery counters on one dashboard |
|
||||
| `webhooker_delivery_attempts_total` | counter | Delivery attempts actually dispatched to a target. A delivery a circuit breaker refused is not one: it is counted as a retry instead, but only when the refusal moves it into `retrying`. An attempt cut off by a stop is not counted (see Shutdown) |
|
||||
| `webhooker_delivery_attempts_total` | counter | Delivery attempts actually dispatched to a target. A delivery a circuit breaker refused is not one: it is counted as a retry instead, but only when the refusal moves it into `retrying` |
|
||||
| `webhooker_deliveries_succeeded_total` | counter | Deliveries that reached `delivered` |
|
||||
| `webhooker_deliveries_failed_total` | counter | Deliveries that failed terminally and will not be retried |
|
||||
| `webhooker_delivery_retries_total` | counter | Deliveries put back into `retrying` |
|
||||
@@ -3290,8 +3284,7 @@ stop-hook log:
|
||||
3. `server` — the HTTP drain, bounded by `server.ShutdownTimeout` (**3
|
||||
seconds**) and by what the hooks before it left, then a Sentry flush if
|
||||
`SENTRY_DSN` is set
|
||||
4. `delivery.Engine` — cuts off any `http` or `slack` request in flight, waits
|
||||
for its workers, then closes the archive databases
|
||||
4. `delivery.Engine` — waits for its workers, then closes the archive databases
|
||||
5. `healthcheck`
|
||||
6. `WebhookDBManager`
|
||||
7. the database close
|
||||
@@ -3302,12 +3295,6 @@ by the stop context, so a wedge there consumes the 5 seconds before the HTTP
|
||||
server hook is ever entered. The hooks after the server are microsecond-scale in
|
||||
normal operation.
|
||||
|
||||
The delivery engine cancels a delivery request in flight. That attempt is not
|
||||
recorded or counted in the metrics, and the delivery keeps its `pending` or
|
||||
`retrying` status, so startup recovery sends it again on the next start (see
|
||||
Recovery paths under Bounded Worker Pool). If the receiver already had it, it
|
||||
receives it twice: the at-least-once duplicate described under Rationale.
|
||||
|
||||
The HTTP drain budget is deliberately **shorter** than the sequence budget. Were
|
||||
the two equal, a drain that used its whole budget would exhaust the sequence
|
||||
budget at the instant it finished, and every later hook — the delivery engine,
|
||||
|
||||
@@ -2,16 +2,11 @@ package delivery_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/fx"
|
||||
"gorm.io/gorm/clause"
|
||||
@@ -277,154 +272,6 @@ func TestEngine_StopHookHonoursStopTimeout(t *testing.T) {
|
||||
requireStopHookExpires(t, lc.hooks[0], "delivery engine")
|
||||
}
|
||||
|
||||
// seedHeldDelivery gives the setup's webhook an http target with
|
||||
// maxRetries, pointed at a receiver that keeps every request open until
|
||||
// the client gives up on it, and seeds one delivery to that target at
|
||||
// status. It returns the task that sends attempt attemptNum of the
|
||||
// delivery, and a channel that is closed when a request reaches the
|
||||
// receiver.
|
||||
func seedHeldDelivery(
|
||||
t *testing.T,
|
||||
s iSetup,
|
||||
maxRetries int,
|
||||
status database.DeliveryStatus,
|
||||
attemptNum int,
|
||||
) (delivery.Task, <-chan struct{}) {
|
||||
t.Helper()
|
||||
|
||||
arrived := make(chan struct{})
|
||||
|
||||
var once sync.Once
|
||||
|
||||
ts := httptest.NewServer(http.HandlerFunc(
|
||||
func(_ http.ResponseWriter, r *http.Request) {
|
||||
// The server notices the client hanging up only once the
|
||||
// body has been read; until then r.Context() never ends.
|
||||
_, _ = io.Copy(io.Discard, r.Body)
|
||||
|
||||
once.Do(func() { close(arrived) })
|
||||
|
||||
<-r.Context().Done()
|
||||
},
|
||||
))
|
||||
t.Cleanup(ts.Close)
|
||||
|
||||
cfg := iHTTPConfig(ts.URL)
|
||||
targetID := uuid.New().String()
|
||||
|
||||
iCreateWebhook(t, s.MainDB, s.WebhookID, "hook")
|
||||
iCreateTarget(
|
||||
t, s.MainDB, targetID, s.WebhookID, "held",
|
||||
database.TargetTypeHTTP, cfg, maxRetries,
|
||||
)
|
||||
|
||||
event := iSeedEvent(t, s.WebhookDB, s.WebhookID, `{"held":true}`)
|
||||
d := iSeedDelivery(t, s.WebhookDB, event.ID, targetID, status)
|
||||
|
||||
bodyStr := event.Body
|
||||
|
||||
return iTask(
|
||||
d, event, s.WebhookID, targetID, "held", cfg,
|
||||
maxRetries, attemptNum, &bodyStr,
|
||||
), arrived
|
||||
}
|
||||
|
||||
// stopDuringAttempt waits until the receiver holds an attempt open, then
|
||||
// stops the engine through its real stop hook.
|
||||
func stopDuringAttempt(
|
||||
t *testing.T, lc *recordingLifecycle, arrived <-chan struct{},
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
select {
|
||||
case <-arrived:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the delivery attempt never reached the receiver")
|
||||
}
|
||||
|
||||
require.NoError(t, lc.hooks[0].OnStop(context.Background()))
|
||||
}
|
||||
|
||||
// requireRequeued runs restart recovery on the stopped engine and
|
||||
// requires it to queue task's attempt again on ch.
|
||||
func requireRequeued(
|
||||
t *testing.T, s iSetup, ch <-chan delivery.Task, task delivery.Task,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
s.Engine.ExportRecoverWebhookDeliveries(
|
||||
context.Background(), s.WebhookID,
|
||||
)
|
||||
|
||||
select {
|
||||
case got := <-ch:
|
||||
assert.Equal(t, task.DeliveryID, got.DeliveryID)
|
||||
assert.Equal(t, task.AttemptNum, got.AttemptNum)
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("restart recovery did not queue the delivery again")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEngine_StopLeavesFireAndForgetAttemptPending is the regression
|
||||
// test for a stop that turned an attempt in flight into a terminal
|
||||
// failure. Stopping the engine cancels the request; for a target with
|
||||
// no retries that was the delivery's only attempt, recorded as failed
|
||||
// with "context canceled", and nothing ever sent it again. The
|
||||
// delivery must instead stay pending, with no result and no count in
|
||||
// the attempt metrics, for restart recovery to send.
|
||||
func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newISetup(t)
|
||||
reg := mIsolate(t, s)
|
||||
lc := startEngineViaHook(t, s.Engine)
|
||||
|
||||
task, arrived := seedHeldDelivery(
|
||||
t, s, 0, database.DeliveryStatusPending, 1,
|
||||
)
|
||||
|
||||
s.Engine.Notify([]delivery.Task{task})
|
||||
stopDuringAttempt(t, lc, arrived)
|
||||
|
||||
iAssertStatus(
|
||||
t, s.WebhookDB, task.DeliveryID, database.DeliveryStatusPending,
|
||||
)
|
||||
assert.Empty(t, iResults(t, s.WebhookDB, task.DeliveryID))
|
||||
assert.Zero(t, mCounter(t, reg, mAttempts, mTypeHTTP))
|
||||
assert.Zero(t, mHTTPDurations(t, reg))
|
||||
|
||||
requireRequeued(t, s, s.Engine.ExportDeliveryCh(), task)
|
||||
}
|
||||
|
||||
// TestEngine_StopLeavesLastRetryAttemptRetrying is the same regression
|
||||
// for a target with retries, on its last attempt: the cancelled attempt
|
||||
// used to exhaust the retries and end the delivery failed. It must stay
|
||||
// retrying with only its earlier attempt recorded, and restart recovery
|
||||
// must send that last attempt again.
|
||||
func TestEngine_StopLeavesLastRetryAttemptRetrying(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const maxRetries = 2
|
||||
|
||||
s := newISetup(t)
|
||||
lc := startEngineViaHook(t, s.Engine)
|
||||
|
||||
task, arrived := seedHeldDelivery(
|
||||
t, s, maxRetries, database.DeliveryStatusRetrying, maxRetries,
|
||||
)
|
||||
iSeedFailedResult(t, s.WebhookDB, task.DeliveryID)
|
||||
|
||||
s.Engine.ExportScheduleRetry(task, 0)
|
||||
stopDuringAttempt(t, lc, arrived)
|
||||
|
||||
iAssertStatus(
|
||||
t, s.WebhookDB, task.DeliveryID, database.DeliveryStatusRetrying,
|
||||
)
|
||||
assert.Len(t, iResults(t, s.WebhookDB, task.DeliveryID), 1)
|
||||
|
||||
requireRequeued(t, s, s.Engine.ExportRetryCh(), task)
|
||||
}
|
||||
|
||||
// deliverToArchive gives the setup's webhook a database target,
|
||||
// runs one delivery to it through the running engine, and returns
|
||||
// the target's ID and archive file path. The archive writer holds
|
||||
|
||||
@@ -32,8 +32,7 @@ type Scheduler interface {
|
||||
// Engine.observeAttempt, alongside the DeliveryResult it records for
|
||||
// it. Deliver is also entered for attempts that never happen — an
|
||||
// open circuit breaker refuses one — so the count cannot be taken
|
||||
// from around this call. An attempt cut off by a stop is neither
|
||||
// counted nor recorded.
|
||||
// from around this call.
|
||||
type Target interface {
|
||||
Deliver(
|
||||
ctx context.Context,
|
||||
|
||||
@@ -57,11 +57,7 @@ type httpCore struct {
|
||||
// recorded and no circuit breaker is consulted. A positive
|
||||
// maxRetries gates the attempt on the circuit breaker and
|
||||
// schedules a backed-off retry on failure.
|
||||
//
|
||||
// ctx is the context attempt sends its request with. Stopping the
|
||||
// engine cancels it; see interruptedByStop.
|
||||
func (c *httpCore) deliver(
|
||||
ctx context.Context,
|
||||
webhookDB *gorm.DB,
|
||||
d *database.Delivery,
|
||||
task *Task,
|
||||
@@ -70,55 +66,21 @@ func (c *httpCore) deliver(
|
||||
attempt func() attemptResult,
|
||||
) {
|
||||
if maxRetries == 0 {
|
||||
c.fireAndForget(ctx, webhookDB, d, attempt())
|
||||
c.fireAndForget(webhookDB, d, attempt())
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
c.withRetry(
|
||||
ctx, webhookDB, d, task, sched, maxRetries, attempt,
|
||||
webhookDB, d, task, sched, maxRetries, attempt,
|
||||
)
|
||||
}
|
||||
|
||||
// interruptedByStop reports whether an attempt failed because the
|
||||
// engine is stopping, which cancels the context its request is made
|
||||
// with. Such an attempt is not recorded: no DeliveryResult, no circuit
|
||||
// breaker outcome, no status change, no count in the attempt metrics.
|
||||
// The delivery stays pending or retrying, and startup recovery sends
|
||||
// it again on the next start.
|
||||
//
|
||||
// A target's own timeout leaves that context alone, so its attempt is
|
||||
// still recorded as the failure it is.
|
||||
func (c *httpCore) interruptedByStop(
|
||||
ctx context.Context,
|
||||
d *database.Delivery,
|
||||
res attemptResult,
|
||||
) bool {
|
||||
if res.success || ctx.Err() == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
c.eng.log.Warn(
|
||||
"delivery attempt interrupted by shutdown; not recorded, "+
|
||||
"it is sent again on the next start",
|
||||
"delivery_id", d.ID,
|
||||
"event_id", d.EventID,
|
||||
"target_id", d.TargetID,
|
||||
)
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *httpCore) fireAndForget(
|
||||
ctx context.Context,
|
||||
webhookDB *gorm.DB,
|
||||
d *database.Delivery,
|
||||
res attemptResult,
|
||||
) {
|
||||
if c.interruptedByStop(ctx, d, res) {
|
||||
return
|
||||
}
|
||||
|
||||
c.eng.observeAttempt(d.Target.Type, res.elapsed())
|
||||
|
||||
err := c.eng.recordResult(
|
||||
@@ -148,7 +110,6 @@ func (c *httpCore) fireAndForget(
|
||||
}
|
||||
|
||||
func (c *httpCore) withRetry(
|
||||
ctx context.Context,
|
||||
webhookDB *gorm.DB,
|
||||
d *database.Delivery,
|
||||
task *Task,
|
||||
@@ -170,10 +131,6 @@ func (c *httpCore) withRetry(
|
||||
|
||||
res := attempt()
|
||||
|
||||
if c.interruptedByStop(ctx, d, res) {
|
||||
return
|
||||
}
|
||||
|
||||
c.eng.observeAttempt(d.Target.Type, res.elapsed())
|
||||
|
||||
err := c.eng.recordResult(
|
||||
@@ -438,7 +395,7 @@ func (t *httpTarget) Deliver(
|
||||
}
|
||||
|
||||
t.deliver(
|
||||
ctx, webhookDB, d, task, sched,
|
||||
webhookDB, d, task, sched,
|
||||
d.Target.MaxRetries, attempt,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -82,7 +82,7 @@ func (t *slackTarget) Deliver(
|
||||
}
|
||||
|
||||
t.deliver(
|
||||
ctx, webhookDB, d, task, sched,
|
||||
webhookDB, d, task, sched,
|
||||
d.Target.MaxRetries, attempt,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -5,7 +5,6 @@ import (
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -26,11 +25,6 @@ const (
|
||||
// its network namespace even with a published port.
|
||||
wildcardV4 = "0.0.0.0"
|
||||
|
||||
// loopbackV6 and wildcardV6 are the IPv6 counterparts of the two
|
||||
// above.
|
||||
loopbackV6 = "::1"
|
||||
wildcardV6 = "::"
|
||||
|
||||
// unavailableAddr is a TEST-NET-1 address (RFC 5737). It is a
|
||||
// well-formed literal that no host is assigned, so binding it
|
||||
// fails with EADDRNOTAVAIL rather than succeeding somewhere
|
||||
@@ -122,24 +116,6 @@ func otherLocalAddr(t *testing.T) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// skipWithoutIPv6 skips the test when this host cannot bind
|
||||
// loopbackV6. Without IPv6 there is no second family for a bind to
|
||||
// claim or leave alone.
|
||||
func skipWithoutIPv6(t *testing.T) {
|
||||
t.Helper()
|
||||
|
||||
var listenCfg net.ListenConfig
|
||||
|
||||
l, err := listenCfg.Listen(
|
||||
t.Context(), "tcp6", net.JoinHostPort(loopbackV6, "0"),
|
||||
)
|
||||
if err != nil {
|
||||
t.Skip("host has no IPv6 loopback to bind")
|
||||
}
|
||||
|
||||
require.NoError(t, l.Close())
|
||||
}
|
||||
|
||||
// startBoundServer starts the wired app with the given bind address
|
||||
// on a free port and returns that port. The app is stopped on
|
||||
// cleanup.
|
||||
@@ -316,50 +292,6 @@ func TestBindAddress_WildcardReachesOtherAddresses(t *testing.T) {
|
||||
requireDialable(t, net.JoinHostPort(other, strconv.Itoa(port)))
|
||||
}
|
||||
|
||||
// TestBindAddress_WildcardV4IsIPv4Only pins that 0.0.0.0 answers on
|
||||
// IPv4 alone, as the README says. Go listens on 0.0.0.0 with a socket
|
||||
// that answers IPv6 too unless told otherwise, which put the admin UI
|
||||
// and the receiver on every IPv6 address of a host whose operator had
|
||||
// chosen an IPv4 bind.
|
||||
func TestBindAddress_WildcardV4IsIPv4Only(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
skipWithoutIPv6(t)
|
||||
|
||||
port := startBoundServer(t, wildcardV4)
|
||||
|
||||
// Positive control: the service really is up, on IPv4.
|
||||
requireDialable(t, net.JoinHostPort(loopbackV4, strconv.Itoa(port)))
|
||||
|
||||
dialer := net.Dialer{Timeout: dialTimeout}
|
||||
|
||||
conn, err := dialer.DialContext(
|
||||
t.Context(), "tcp",
|
||||
net.JoinHostPort(loopbackV6, strconv.Itoa(port)),
|
||||
)
|
||||
if err == nil {
|
||||
_ = conn.Close()
|
||||
}
|
||||
|
||||
require.ErrorIs(
|
||||
t, err, syscall.ECONNREFUSED,
|
||||
"a %s bind answered on %s", wildcardV4, loopbackV6,
|
||||
)
|
||||
}
|
||||
|
||||
// TestBindAddress_WildcardV6ReachesBothFamilies is the counterpart:
|
||||
// :: is the value that binds IPv6, and it answers IPv4 as well.
|
||||
func TestBindAddress_WildcardV6ReachesBothFamilies(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
skipWithoutIPv6(t)
|
||||
|
||||
port := startBoundServer(t, wildcardV6)
|
||||
|
||||
requireDialable(t, net.JoinHostPort(loopbackV6, strconv.Itoa(port)))
|
||||
requireDialable(t, net.JoinHostPort(loopbackV4, strconv.Itoa(port)))
|
||||
}
|
||||
|
||||
// TestBindAddress_ServesRequestsOnConfiguredAddress proves the bound
|
||||
// listener serves the application rather than merely accepting TCP,
|
||||
// so a bind address that is honoured cannot be mistaken for one that
|
||||
|
||||
+1
-28
@@ -1,11 +1,9 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
@@ -41,21 +39,6 @@ func (s *Server) listenAddr() string {
|
||||
)
|
||||
}
|
||||
|
||||
// listenNetwork returns the network the HTTP listener binds on, "tcp4"
|
||||
// for an IPv4 BindAddress and "tcp" for any other.
|
||||
//
|
||||
// On "tcp", Go binds 0.0.0.0 with a socket that answers IPv6 as well,
|
||||
// exactly as it binds ::. "tcp4" keeps 0.0.0.0 to IPv4, which is what
|
||||
// the README says it means. :: stays on "tcp" and answers both.
|
||||
func (s *Server) listenNetwork() string {
|
||||
addr, err := netip.ParseAddr(s.params.Config.BindAddress)
|
||||
if err == nil && addr.Is4() {
|
||||
return "tcp4"
|
||||
}
|
||||
|
||||
return "tcp"
|
||||
}
|
||||
|
||||
// newHTTPServer builds the HTTP server for this Server's
|
||||
// configuration.
|
||||
//
|
||||
@@ -84,17 +67,7 @@ func (s *Server) serveUntilShutdown() {
|
||||
|
||||
s.log.Info("http begin listen", "listenaddr", s.httpServer.Addr)
|
||||
|
||||
// ListenAndServe always listens on "tcp", which would undo
|
||||
// listenNetwork.
|
||||
var listenCfg net.ListenConfig
|
||||
|
||||
listener, err := listenCfg.Listen(
|
||||
context.Background(), s.listenNetwork(), s.httpServer.Addr,
|
||||
)
|
||||
if err == nil {
|
||||
err = s.httpServer.Serve(listener)
|
||||
}
|
||||
|
||||
err := s.httpServer.ListenAndServe()
|
||||
if err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
s.log.Error("listen error", "error", err)
|
||||
s.shutdownWithFailure()
|
||||
|
||||
@@ -135,8 +135,6 @@ func New(lc fx.Lifecycle, params ServerParams) (*Server, error) {
|
||||
s.httpServer = s.newHTTPServer()
|
||||
|
||||
lc.Append(fx.Hook{
|
||||
//nolint:contextcheck // The listener outlives this hook, whose
|
||||
// context fx cancels once the app has started.
|
||||
OnStart: func(_ context.Context) error {
|
||||
s.startupTime = time.Now()
|
||||
go s.Run()
|
||||
|
||||
Reference in New Issue
Block a user