amadan

rastrillo / aviso Public

Send: stop aliasing the payload, sanitise transport errors, bound the settle

Codex's review of Send, five findings, all real:

webpush-go pads in place into the payload's spare capacity, so one
payload fanned out to eight devices shared a backing array between
goroutines. Capping the slice's capacity forces each append to
reallocate; the new test fails under -race without it.

Stripping the outer url.Error was not redaction: net/http quotes a
bad Location header verbatim, and a push service could reflect the
endpoint there. Transport errors now collapse to ErrTransport,
ErrDialRefused, or a wrapped context error, with the original
discarded rather than wrapped.

The post-answer store update ran under WithoutCancel with no deadline;
with rastrillo's single writer connection, a held transaction would
pin a semaphore slot and a cancelled Send forever. It now has its own
bound (dbTimeout), and settle is a seam so the cancelled-context and
blocked-writer cases are tested deterministically rather than by
racing HTTP against cancel.

Also: two batches on one Service are tested to share one bound, and
the mid-batch cancellation test accepts an in-flight request that
legitimately completed before the cancellation reached the transport.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 61fc8a905dca1bbe6df374c4c40a4a23821c465c parent 3175bd1
3 files changed, +200 −29
  • aviso.go +5 −0
  • send.go +49 −22
  • send_test.go +146 −7
diff --git a/aviso.go b/aviso.go
index a819ae7..331ee3a 100644
--- a/aviso.go
+++ b/aviso.go
@@ -77,6 +77,10 @@ type Service struct {
client *http.Client // built by newClient (ssrf.go); tests may replace it
sem chan struct{}
now func() time.Time
+ // dbTimeout bounds the store update after a push service answers.
+ // It is independent of the caller's context on purpose (see
+ // settle) and bounded on purpose: the writer is one connection.
+ dbTimeout time.Duration
}
// New validates cfg and returns a ready *Service.
@@ -108,6 +112,7 @@ func New(cfg Config) (*Service, error) {
client: newClient(),
sem: make(chan struct{}, cfg.Concurrency),
now: time.Now,
+ dbTimeout: 5 * time.Second,
}, nil
}
diff --git a/send.go b/send.go
index 82e105f..210fcbc 100644
--- a/send.go
+++ b/send.go
@@ -5,8 +5,8 @@ import (
"errors"
"fmt"
"io"
+ "net"
"net/http"
- "net/url"
"strconv"
"strings"
"sync"
@@ -164,6 +164,11 @@ func (s *Service) sendOne(ctx context.Context, st Stored, payload []byte, o Opti
if urgency == "" {
urgency = webpush.UrgencyNormal
}
+ // webpush-go wraps the payload in a bytes.Buffer and appends the
+ // record delimiter and padding in place. With spare capacity, the
+ // concurrent sends of one batch would write into one backing
+ // array; capping the capacity forces each append to reallocate.
+ payload = payload[:len(payload):len(payload)]
resp, err := webpush.SendNotificationWithContext(ctx, payload,
&webpush.Subscription{Endpoint: st.Endpoint, Keys: webpush.Keys{P256dh: st.P256dh, Auth: st.Auth}},
&webpush.Options{
@@ -183,41 +188,63 @@ func (s *Service) sendOne(ctx context.Context, st Stored, payload []byte, o Opti
// Drain a bounded amount for keep-alive; the body is never read
// into anything a log could see.
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBodyRead))
- res.Status = resp.StatusCode
- // Store updates run under a context the caller's cancellation
- // cannot interrupt: a 410 the service already answered must prune
- // whether or not the batch was cancelled a moment later.
- dbCtx := context.WithoutCancel(ctx)
+ return s.settle(ctx, st, resp.StatusCode, resp.Header.Get("Retry-After"))
+}
+
+// settle turns the push service's answer into a Result and the store
+// update it implies. Store updates run under a context the caller's
+// cancellation cannot interrupt — a 410 the service already answered
+// must prune whether or not the batch was cancelled a moment later —
+// but with their own bound, because the writer is one connection and
+// a transaction holding it must not pin a semaphore slot forever.
+func (s *Service) settle(ctx context.Context, st Stored, status int, retryAfter string) Result {
+ res := Result{ID: st.ID, Status: status}
+ dbCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), s.dbTimeout)
+ defer cancel()
switch {
- case resp.StatusCode >= 200 && resp.StatusCode < 300:
+ case status >= 200 && status < 300:
if err := s.confirm(dbCtx, st.ID, st.Revision); err != nil {
s.cfg.Logger.Warn("aviso: confirm failed", "id", st.ID, "err", err)
}
- case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone:
- res.Err = fmt.Errorf("aviso: push service says subscription gone (%d)", resp.StatusCode)
+ case status == http.StatusNotFound || status == http.StatusGone:
+ res.Err = fmt.Errorf("aviso: push service says subscription gone (%d)", status)
if err := s.prune(dbCtx, st.ID, st.Revision); err != nil {
s.cfg.Logger.Warn("aviso: prune failed", "id", st.ID, "err", err)
}
- case resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode == http.StatusServiceUnavailable:
- res.RetryAfter = parseRetryAfter(resp.Header.Get("Retry-After"), s.now())
- res.Err = fmt.Errorf("aviso: push service throttled (%d)", resp.StatusCode)
+ case status == http.StatusTooManyRequests || status == http.StatusServiceUnavailable:
+ res.RetryAfter = parseRetryAfter(retryAfter, s.now())
+ res.Err = fmt.Errorf("aviso: push service throttled (%d)", status)
default:
- res.Err = fmt.Errorf("aviso: push service refused (%d)", resp.StatusCode)
+ res.Err = fmt.Errorf("aviso: push service refused (%d)", status)
}
return res
}
-// redact strips the request URL from a transport error: *url.Error
-// prints it, and the endpoint is the one secret in this package that
-// would otherwise reach a log. What survives is the operation and the
-// underlying cause (a dial refusal, a timeout), which name at most the
-// push service's IP, never the endpoint's path.
+// ErrTransport is the sanitised form of any transport failure that is
+// not a cancellation, a timeout or the guard's own refusal. The
+// original is discarded, not wrapped: net/http's errors quote the
+// request URL and even a bad Location header verbatim, and the
+// endpoint is the one secret in this package that must never reach a
+// log.
+var ErrTransport = errors.New("aviso: transport error")
+
+// ErrDialRefused is the guard's refusal, surfaced without the address.
+var ErrDialRefused = errors.New("aviso: dial refused by the SSRF guard")
+
func redact(err error) error {
- var ue *url.Error
- if errors.As(err, &ue) {
- return fmt.Errorf("aviso: %s: %w", ue.Op, ue.Err)
+ switch {
+ case errors.Is(err, context.Canceled):
+ return fmt.Errorf("aviso: send: %w", context.Canceled)
+ case errors.Is(err, context.DeadlineExceeded):
+ return fmt.Errorf("aviso: send: %w", context.DeadlineExceeded)
+ case strings.Contains(err.Error(), "aviso: dial refused"):
+ return ErrDialRefused
+ }
+ var ne net.Error
+ if errors.As(err, &ne) && ne.Timeout() {
+ return fmt.Errorf("aviso: send: %w", context.DeadlineExceeded)
}
- return fmt.Errorf("aviso: send: %w", err)
+ return ErrTransport
}
func parseRetryAfter(v string, now time.Time) time.Duration {
diff --git a/send_test.go b/send_test.go
index 54c80a5..2074ab7 100644
--- a/send_test.go
+++ b/send_test.go
@@ -276,25 +276,164 @@ func TestSendCancelledMidBatchWaitsForStartedGoroutines(t *testing.T) {
if !errors.Is(err, context.Canceled) {
t.Fatalf("batch err = %v", err)
}
- // A row never started carries the bare context error; a row whose
- // request was cut short carries it wrapped by the transport (and
- // was seen by the recorder).
+ // A row never started carries the bare context error. A row that
+ // was in flight either got cut short (wrapped context error) or
+ // legitimately completed before the cancellation reached the
+ // transport — both are "started"; neither is the bare error.
started, skipped := 0, 0
for _, r := range res {
switch {
case r.Err == context.Canceled:
skipped++
- case r.Err != nil:
+ case r.Err != nil || r.Status != 0:
started++
default:
- t.Fatalf("result with no error after cancellation: %+v", r)
+ t.Fatalf("result neither started nor skipped: %+v", r)
}
}
if started != 2 || skipped != 3 {
t.Fatalf("started=%d skipped=%d, want 2/3", started, skipped)
}
- if seen := len(p.hdr) + 2; seen != 2 { // the two we drained; nothing else reached the service
- t.Fatalf("recorder saw %d requests, want 2", seen)
+ if extra := len(p.hdr); extra != 0 { // the two we drained were the only requests
+ t.Fatalf("recorder saw %d requests beyond the two in flight", extra)
+ }
+}
+
+// Two batches on one Service share one bound: a per-call semaphore
+// would let this reach 6 in flight.
+func TestConcurrencyBoundIsPerService(t *testing.T) {
+ p := newPushRecorder(t)
+ p.hold = make(chan struct{})
+ s := serviceAgainst(t, p)
+ s.sem = make(chan struct{}, 3)
+ ctx := context.Background()
+ for i := 0; i < 6; i++ {
+ _ = s.put(ctx, "alice", sub(p.url("/a"+string(rune('a'+i)))), "")
+ _ = s.put(ctx, "bob", sub(p.url("/b"+string(rune('a'+i)))), "")
+ }
+ done := make(chan struct{}, 2)
+ go func() { _, _ = s.SendTo(ctx, "alice", []byte("x"), Options{}); done <- struct{}{} }()
+ go func() { _, _ = s.SendTo(ctx, "bob", []byte("x"), Options{}); done <- struct{}{} }()
+ for i := 0; i < 3; i++ {
+ <-p.hdr
+ }
+ time.Sleep(50 * time.Millisecond)
+ if got := p.inFlt.Load(); got != 3 {
+ t.Fatalf("in flight across two batches = %d, want 3", got)
+ }
+ close(p.hold)
+ <-done
+ <-done
+ if p.peak.Load() > 3 {
+ t.Fatalf("peak %d exceeded the Service bound", p.peak.Load())
+ }
+}
+
+// webpush-go pads in place into the payload's spare capacity; one
+// payload fanned out to many devices must not share a backing array.
+// -race is what makes this test bite.
+func TestSendDoesNotAliasThePayloadAcrossDevices(t *testing.T) {
+ p := newPushRecorder(t)
+ s := serviceAgainst(t, p)
+ ctx := context.Background()
+ for i := 0; i < 8; i++ {
+ _ = s.put(ctx, "alice", sub(p.url("/"+string(rune('a'+i)))), "")
+ }
+ payload := make([]byte, 1, 4096)
+ payload[0] = 'x'
+ res, err := s.SendTo(ctx, "alice", payload, Options{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, r := range res {
+ if r.Err != nil {
+ t.Fatalf("%+v", r)
+ }
+ }
+ if payload[0] != 'x' || len(payload) != 1 {
+ t.Fatal("caller's payload was modified")
+ }
+}
+
+// A 410 the push service already answered prunes even though the
+// caller's context was cancelled by the time the answer is settled.
+func TestSettlePrunesUnderACancelledContext(t *testing.T) {
+ s := newInternalService(t)
+ ctx, cancel := context.WithCancel(context.Background())
+ _ = s.put(ctx, "alice", sub("https://push.example/one"), "")
+ rows, _ := s.List(ctx, "alice")
+ cancel()
+ res := s.settle(ctx, rows[0], http.StatusGone, "")
+ if res.Status != 410 || res.Err == nil {
+ t.Fatalf("res=%+v", res)
+ }
+ if left, _ := s.List(context.Background(), "alice"); len(left) != 0 {
+ t.Fatal("prune skipped under a cancelled context")
+ }
+ // And the same for confirm.
+ _ = s.put(context.Background(), "alice", sub("https://push.example/two"), "")
+ rows, _ = s.List(context.Background(), "alice")
+ s.now = func() time.Time { return time.Unix(1_800_000_000, 0) }
+ _ = s.settle(ctx, rows[0], http.StatusCreated, "")
+ var confirmed int64
+ _ = s.cfg.DB.QueryRow(`SELECT last_confirmed_at FROM aviso_subscriptions WHERE id=?`, rows[0].ID).Scan(&confirmed)
+ if confirmed != 1_800_000_000 {
+ t.Fatal("confirm skipped under a cancelled context")
+ }
+}
+
+// The store update after an answer has its own bound: a transaction
+// holding the single writer connection must not pin the send forever.
+func TestSettleGivesUpOnABlockedWriter(t *testing.T) {
+ s := newInternalService(t)
+ s.dbTimeout = 100 * time.Millisecond
+ ctx := context.Background()
+ _ = s.put(ctx, "alice", sub("https://push.example/one"), "")
+ rows, _ := s.List(ctx, "alice")
+ tx, err := s.cfg.DB.BeginTx(ctx, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer tx.Rollback()
+ if _, err := tx.Exec(`UPDATE aviso_subscriptions SET revision = revision`); err != nil { // hold the write lock
+ t.Fatal(err)
+ }
+ start := time.Now()
+ done := make(chan Result, 1)
+ go func() { done <- s.settle(ctx, rows[0], http.StatusGone, "") }()
+ select {
+ case res := <-done:
+ if time.Since(start) > 2*time.Second {
+ t.Fatalf("settle took %v with a 100ms bound", time.Since(start))
+ }
+ if res.Status != 410 {
+ t.Fatalf("res=%+v", res)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("settle blocked on the held writer")
+ }
+}
+
+// A 3xx with an unparseable Location makes net/http quote the header
+// verbatim in its error, and a push service could reflect the endpoint
+// there. Nothing of it may survive into the Result.
+func TestSendRedactsMalformedRedirects(t *testing.T) {
+ p := newPushRecorder(t)
+ srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Location", "https://push.example/secret-path/%zz")
+ w.WriteHeader(http.StatusFound)
+ }))
+ defer srv.Close()
+ p.srv = srv
+ s := serviceAgainst(t, p)
+ ctx := context.Background()
+ _ = s.put(ctx, "alice", sub(p.url("/secret-path")), "")
+ res, _ := s.SendTo(ctx, "alice", []byte("x"), Options{})
+ if res[0].Err == nil || strings.Contains(res[0].Err.Error(), "secret") || strings.Contains(res[0].Err.Error(), "zz") {
+ t.Fatalf("error leaks the redirect: %v", res[0].Err)
+ }
+ if !errors.Is(res[0].Err, ErrTransport) {
+ t.Fatalf("want ErrTransport, got %v", res[0].Err)
}
}