diff --git a/internal/forwarder/gate.go b/internal/forwarder/gate.go index ede76b5..e9e765d 100644 --- a/internal/forwarder/gate.go +++ b/internal/forwarder/gate.go @@ -36,15 +36,12 @@ func (g *Gate) Acquire(eventID string) (release func(), wait <-chan struct{}) { ch := make(chan struct{}) g.in[eventID] = ch g.mu.Unlock() - var once sync.Once - return func() { - once.Do(func() { - g.mu.Lock() - delete(g.in, eventID) - g.mu.Unlock() - close(ch) - }) - }, nil + return sync.OnceFunc(func() { + g.mu.Lock() + defer g.mu.Unlock() + delete(g.in, eventID) + close(ch) + }), nil } // InFlight reports how many events are currently gated (tests/metrics). diff --git a/internal/server/forward_audit_test.go b/internal/server/forward_audit_test.go index ddacbec..a28ee70 100644 --- a/internal/server/forward_audit_test.go +++ b/internal/server/forward_audit_test.go @@ -95,8 +95,17 @@ func TestForwardAuditByteIdenticalAndMinimal(t *testing.T) { t.Fatalf("unexpected forwarded header %q", k) } } - if out.Header.Get("X-SubnetLens-Event-ID") != id || out.Header.Get("X-SubnetLens-Signature-256") == "" { - t.Fatalf("signature headers not verbatim: %v", out.Header) + for _, name := range []string{ + "X-SubnetLens-Event-ID", "X-SubnetLens-Event-Key", "X-SubnetLens-Event-Type", + "X-SubnetLens-Signature-Timestamp", "X-SubnetLens-Signature-256", "Content-Type", + } { + want := req.Header.Get(name) + if want == "" { + t.Fatalf("test request lacks %s", name) + } + if got := out.Header.Get(name); got != want { + t.Errorf("forwarded %s = %q, want %q", name, got, want) + } } // Nothing local is folded into the relay body: the forwarded bytes // equal the producer submission, which carried no decision/rule diff --git a/internal/server/server.go b/internal/server/server.go index e0acdf2..127a97d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -5,7 +5,8 @@ // 202 accepted (persisted; forwarded in proxy mode), 401 bad signature or // timestamp, 400 malformed, 413 oversize, 405 non-POST, 503 saturated or // store unavailable with Retry-After, 502 ScanRelay delivery failed. -// A duplicate host event is idempotent (202 when forwarded, 502 when not). +// A duplicate host event keeps the first stored payload. In proxy mode it +// returns 202 once forwarded and 502 if delivery has not succeeded. // H1 is persisted separately and never forwarded; identical retries are 202, // while conflicting reuse of a heartbeat ID is 409. package server @@ -327,7 +328,8 @@ func (s *Server) persistAndReply(ctx context.Context, w http.ResponseWriter, bod pctx, cancel := context.WithTimeout(ctx, s.cfg.PersistBudget) defer cancel() - headersJSON, err := json.Marshal(store.WhitelistForwardHeaders(headerMap(r.Header))) + headers := store.WhitelistForwardHeaders(headerMap(r.Header)) + headersJSON, err := json.Marshal(headers) if err != nil { return errMalformedBody } @@ -374,11 +376,15 @@ func (s *Server) persistAndReply(ctx context.Context, w http.ResponseWriter, bod // doing its job. In-flight duplicates never reach here: // they wait on the gate and follow the stored state. s.Health.AddDuplicate() - if fwd, _, _, ferr := s.events.ForwardState(ctx, eventID); ferr == nil && fwd { + if s.cfg.Mode == config.ModeTap { writeAccepted(w, eventID, eventType) return nil } - if s.cfg.Mode == config.ModeTap || s.forwardOne(ctx, body, headerMap(r.Header), eventID) { + forwarded, samePayload, err := s.events.RetryForwardState(ctx, eventID, body) + if err != nil { + return err + } + if forwarded || (samePayload && s.forwardOne(ctx, body, headers, eventID)) { writeAccepted(w, eventID, eventType) return nil } @@ -391,7 +397,7 @@ func (s *Server) persistAndReply(ctx context.Context, w http.ResponseWriter, bod writeAccepted(w, eventID, eventType) return nil } - if s.forwardOne(ctx, body, headerMap(r.Header), eventID) { + if s.forwardOne(ctx, body, headers, eventID) { s.Health.AddAccepted() writeAccepted(w, eventID, eventType) return nil @@ -413,13 +419,16 @@ func (s *Server) reportPersistError(w http.ResponseWriter, eventID string, err e s.writeSaturated(w, "persist unavailable") } -// replyForStoredState answers a duplicate from the forwarded flag: 202 when -// the event landed at ScanRelay, 502 when it did not so Pro retries. It never -// forwards inline — a second forward inside this deadline could exceed it. +// replyForStoredState accepts a stored tap event or a forwarded proxy event. +// A waiter never forwards inline; a new request owns the next retry. func (s *Server) replyForStoredState(ctx context.Context, w http.ResponseWriter, eventID, eventType string) { s.Health.AddDuplicate() forwarded, _, _, err := s.events.ForwardState(ctx, eventID) - if err != nil || !forwarded { + if err != nil { + s.writeSaturated(w, "persist unavailable") + return + } + if s.cfg.Mode == config.ModeProxy && !forwarded { writeError(w, http.StatusBadGateway, "scanrelay delivery failed") return } diff --git a/internal/server/server_test.go b/internal/server/server_test.go index a36c99a..0d75c3b 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -221,16 +221,17 @@ func TestContractTable(t *testing.T) { // 502: ScanRelay down. h.relay.mode.Store("fail") failID := uniqueID() - if rec := h.signedPost(t, failID, "host-b", "host.appeared", now); rec.Code != http.StatusBadGateway { + body := eventBody(failID, "host-b", "host.appeared") + sig := sign(h.secret, rts, failID, "host-b", "host.appeared", body) + if rec := h.post(t, body, failID, "host-b", "host.appeared", now, sig); rec.Code != http.StatusBadGateway { t.Fatalf("forward failure: got %d", rec.Code) } - // 202 duplicate of an unforwarded event: waiter path returns 502, and - // the retry after a successful forward returns 202. - if rec := h.signedPost(t, failID, "host-b", "host.appeared", now); rec.Code != http.StatusBadGateway { + // A retry remains 502 until the relay accepts the stored payload. + if rec := h.post(t, body, failID, "host-b", "host.appeared", now, sig); rec.Code != http.StatusBadGateway { t.Fatalf("duplicate of unforwarded: got %d", rec.Code) } h.relay.mode.Store("accept") - if rec := h.signedPost(t, failID, "host-b", "host.appeared", now); rec.Code != http.StatusAccepted { + if rec := h.post(t, body, failID, "host-b", "host.appeared", now, sig); rec.Code != http.StatusAccepted { t.Fatalf("retry after failure: got %d %s", rec.Code, rec.Body.String()) } if h.srv.gate.InFlight() != 0 { @@ -344,11 +345,63 @@ func TestWaiterReturns502WhenWinnerFails(t *testing.T) { } // Pro's retry then succeeds and is accepted. h.relay.mode.Store("accept") - if rec := h.signedPost(t, id, "host-wf", "host.appeared", now); rec.Code != http.StatusAccepted { + if rec := h.post(t, body, id, "host-wf", "host.appeared", now, sig); rec.Code != http.StatusAccepted { t.Fatalf("retry: got %d", rec.Code) } } +func TestProxyRetryCannotReplaceStoredPayload(t *testing.T) { + h := newHarness(t, nil) + now := time.Now() + id, key, typ := uniqueID(), "host-retry", "host.appeared" + body := eventBody(id, key, typ) + post := func(payload []byte, at time.Time) int { + sig := sign(h.secret, strconv.FormatInt(at.Unix(), 10), id, key, typ, payload) + return h.post(t, payload, id, key, typ, at, sig).Code + } + h.relay.mode.Store("fail") + if got := post(body, now); got != http.StatusBadGateway { + t.Fatalf("initial delivery status = %d", got) + } + h.relay.mode.Store("accept") + retryAt := now.Add(time.Second) + changed := bytes.Replace(body, []byte(`"cycle":1`), []byte(`"cycle":2`), 1) + if got := post(changed, retryAt); got != http.StatusBadGateway || h.relay.hits.Load() != 1 { + t.Fatalf("conflicting retry status = %d, relay hits = %d", got, h.relay.hits.Load()) + } + if got := post(body, retryAt); got != http.StatusAccepted || h.relay.hits.Load() != 2 { + t.Fatalf("exact retry status = %d, relay hits = %d", got, h.relay.hits.Load()) + } + if got := post(changed, retryAt); got != http.StatusAccepted || h.relay.hits.Load() != 2 { + t.Fatalf("conflict after delivery status = %d, relay hits = %d", got, h.relay.hits.Load()) + } + stored, err := h.srv.events.StoredEvent(context.Background(), id) + h.relay.mu.Lock() + defer h.relay.mu.Unlock() + if err != nil || !bytes.Equal(stored.Payload, body) || !bytes.Equal(h.relay.bodies[1], body) { + t.Fatalf("stored and forwarded payloads must match the original: %v", err) + } +} + +func TestTapWaiterFollowsStoredEvent(t *testing.T) { + h := newHarness(t, func(c *config.Config) { c.Mode = config.ModeTap }) + ctx := context.Background() + id := uniqueID() + rec := httptest.NewRecorder() + h.srv.replyForStoredState(ctx, rec, id, "host.appeared") + if rec.Code != http.StatusServiceUnavailable || rec.Header().Get("Retry-After") == "" { + t.Fatalf("missing event status = %d, Retry-After = %q", rec.Code, rec.Header().Get("Retry-After")) + } + if got := h.signedPost(t, id, "host-tap", "host.appeared", time.Now()).Code; got != http.StatusAccepted { + t.Fatalf("initial delivery status = %d", got) + } + rec = httptest.NewRecorder() + h.srv.replyForStoredState(ctx, rec, id, "host.appeared") + if rec.Code != http.StatusAccepted || h.relay.hits.Load() != 0 { + t.Fatalf("tap waiter status = %d, relay hits = %d", rec.Code, h.relay.hits.Load()) + } +} + func TestCancelDuringPersistThenRetry(t *testing.T) { h := newHarness(t, func(c *config.Config) { c.Mode = config.ModeTap }) now := time.Now() diff --git a/internal/server/sweep.go b/internal/server/sweep.go index 13c709b..6473575 100644 --- a/internal/server/sweep.go +++ b/internal/server/sweep.go @@ -84,7 +84,11 @@ func (s *Server) sweepOne(ctx context.Context, ev store.RawEvent) (done, landed if s.now().Sub(ev.SignatureTimestamp) >= s.cfg.TimestampTolerance-s.cfg.SweepSafetyMargin { return s.park(ctx, ev.EventID, "stale_for_sweep"), false } - if fwd, _, _, err := s.events.ForwardState(ctx, ev.EventID); err != nil || fwd { + fwd, _, _, err := s.events.ForwardState(ctx, ev.EventID) + if err != nil { + return false, false + } + if fwd { return true, false } var headers map[string]string diff --git a/internal/store/forwarding.go b/internal/store/forwarding.go index f67faee..c4daa5f 100644 --- a/internal/store/forwarding.go +++ b/internal/store/forwarding.go @@ -18,6 +18,14 @@ func (s *Store) ForwardState(ctx context.Context, eventID string) (forwarded boo return fwd == 1, attempts, last.String, err } +// RetryForwardState checks whether the event already landed or a newly +// signed retry carries the exact bytes originally accepted for this ID. +func (s *Store) RetryForwardState(ctx context.Context, eventID string, payload []byte) (forwarded, samePayload bool, err error) { + err = s.db.QueryRowContext(ctx, `SELECT forwarded, payload = ? FROM raw_events WHERE event_id = ?`, payload, eventID). + Scan(&forwarded, &samePayload) + return forwarded, samePayload, err +} + // MarkForwarded records a successful delivery once; repeats are no-ops. func (s *Store) MarkForwarded(ctx context.Context, eventID string, at time.Time) (bool, error) { // returns false on any error (captured var is only read on success) matched := false diff --git a/internal/store/retention.go b/internal/store/retention.go index e10fb0e..f3f56c7 100644 --- a/internal/store/retention.go +++ b/internal/store/retention.go @@ -49,7 +49,7 @@ func (s *Store) Checkpoint(ctx context.Context) error { } return nil } - return s.enqueue(ctx, job) + return s.enqueue(job) } // MarkUndeliverable parks an event the forwarder can never send (e.g. its diff --git a/internal/store/store.go b/internal/store/store.go index 8804305..9989cf6 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -7,6 +7,7 @@ package store import ( "context" "database/sql" + "errors" "fmt" "net/url" "os" @@ -26,16 +27,14 @@ type writeJob struct { // raw runs on the writer goroutine without opening a transaction, // for statements SQLite forbids inside one (e.g. wal_checkpoint). raw func() error - done chan error + done chan error // buffered so the writer can reply after caller cancellation } -// ErrStoreClosed is returned when the store is shut down; ErrQueueSaturated -// when the bounded writer queue is full. Both mean "retry later", never -// "bad request". -var ( - ErrStoreClosed = fmt.Errorf("store is closed") - ErrQueueSaturated = fmt.Errorf("writer queue saturated") -) +// ErrStoreClosed means the writer rejected the job before running it. +var ErrStoreClosed = errors.New("store is closed") + +// ErrQueueSaturated means the writer queue was full and did not admit the job. +var ErrQueueSaturated = errors.New("writer queue saturated") // DefaultQueueCap bounds the writer channel when the caller has no number. const DefaultQueueCap = 64 @@ -52,12 +51,11 @@ func WithQueueCap(n int) Option { return func(o *options) { o.queueCap = n } } // Store owns the database. The zero value is invalid; use Open. type Store struct { db *sql.DB - jobs chan writeJob // never closed; quit signals shutdown - queueCap int // jobs capacity; the purge loop yields past half + jobs chan writeJob // closed only under submitMu + submitMu sync.Mutex // serializes submission with closing jobs quit chan struct{} wg sync.WaitGroup - closeMu sync.Mutex - closed bool + closeOnce sync.Once closeErr error groupSeq atomic.Int64 decisionSeq atomic.Int64 @@ -100,7 +98,7 @@ func (s *Store) QueueDepth() int { return len(s.jobs) } // QueueCap reports the writer queue capacity QueueDepth is measured // against. Zero on a Store that bypassed Open, which never purges. -func (s *Store) QueueCap() int { return s.queueCap } +func (s *Store) QueueCap() int { return cap(s.jobs) } func nanos(t time.Time) int64 { return t.UTC().UnixNano() } func fromNanos(n int64) time.Time { @@ -270,7 +268,7 @@ func openStore(path string, opts ...Option) (*Store, error) { db.Close() return nil, fmt.Errorf("tighten database permissions: %w", err) } - s := &Store{db: db, jobs: make(chan writeJob, o.queueCap), queueCap: o.queueCap, quit: make(chan struct{}), now: time.Now} + s := &Store{db: db, jobs: make(chan writeJob, o.queueCap), quit: make(chan struct{}), now: time.Now} s.wg.Add(1) go s.writer() return s, nil @@ -282,21 +280,11 @@ func openStore(path string, opts ...Option) (*Store, error) { // A database newer than this code is refused rather than silently misread. func (s *Store) writer() { defer s.wg.Done() - for { + for job := range s.jobs { select { case <-s.quit: - // Answer every queued job before exiting: a caller with no - // deadline must never hang on a store that will never run - // its write. done is always buffered, so this never blocks. - for { - select { - case job := <-s.jobs: - job.done <- ErrStoreClosed - default: - return - } - } - case job := <-s.jobs: + job.done <- ErrStoreClosed + default: job.done <- s.runJob(job) } } @@ -335,54 +323,54 @@ func (s *Store) runJob(job writeJob) (err error) { // "unknown, retry safely" and rely on idempotency (e.g. // PersistEventAndGroup reports Inserted=false on the retry). func (s *Store) write(ctx context.Context, fn func(*sql.Tx) error) error { - return s.enqueue(ctx, writeJob{ctx: ctx, fn: fn, done: make(chan error, 1)}) + return s.enqueue(writeJob{ctx: ctx, fn: fn, done: make(chan error, 1)}) } -// enqueue submits a job without ever blocking the caller: a full queue -// fails fast so ingest can answer 503 instead of piling up memory. -func (s *Store) enqueue(ctx context.Context, job writeJob) error { +// enqueue waits for the outcome of an admitted job. A full queue fails +// fast so ingest can answer 503 instead of waiting for queue space. +func (s *Store) enqueue(job writeJob) error { + if err := s.submit(job); err != nil { + return err + } + select { + case <-job.ctx.Done(): + return job.ctx.Err() + case err := <-job.done: + return err + } +} + +func (s *Store) submit(job writeJob) error { + s.submitMu.Lock() + defer s.submitMu.Unlock() select { case <-s.quit: return ErrStoreClosed - case <-ctx.Done(): - return ctx.Err() default: } + if err := job.ctx.Err(); err != nil { + return err + } select { - case <-s.quit: - return ErrStoreClosed - case <-ctx.Done(): - return ctx.Err() case s.jobs <- job: + return nil default: return ErrQueueSaturated } - select { - case <-s.quit: - // The writer drained the queue and answered every job it saw; - // a job submitted in the race window gets a closed answer here - // instead of hanging a deadline-less caller forever. - return ErrStoreClosed - case <-ctx.Done(): - return ctx.Err() - case err := <-job.done: - return err - } } // Close shuts the store down once and reports the result to every caller. -// A concurrent second caller blocks on the mutex until the first shutdown -// finishes, then receives the same error — nobody returns early while the -// writer is still draining. +// Concurrent callers wait for the same shutdown result while the writer +// rejects queued jobs and finishes any in-flight work. func (s *Store) Close() error { - s.closeMu.Lock() - defer s.closeMu.Unlock() - if s.closed { - return s.closeErr - } - s.closed = true - close(s.quit) // never close s.jobs: a blocked sender must not panic - s.wg.Wait() - s.closeErr = s.db.Close() + s.closeOnce.Do(func() { + s.submitMu.Lock() + close(s.quit) + close(s.jobs) + s.submitMu.Unlock() + + s.wg.Wait() + s.closeErr = s.db.Close() + }) return s.closeErr } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 28ddfae..5bcba96 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -11,6 +11,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "testing" "time" ) @@ -905,71 +906,124 @@ func TestPurgeFailedGroupEvents(t *testing.T) { // H1: shutdown answers every queued job. A slow write occupies the // writer while Background-context callers queue behind it; Close must // release them all with ErrStoreClosed instead of hanging them forever. -func TestCloseDrainsQueuedJobs(t *testing.T) { +func TestCloseRejectsQueuedJobs(t *testing.T) { s, err := Open(filepath.Join(t.TempDir(), "drain.db"), WithQueueCap(16)) if err != nil { t.Fatal(err) } + defer s.Close() + entered := make(chan struct{}) release := make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() slowDone := make(chan error, 1) go func() { slowDone <- s.write(context.Background(), func(tx *sql.Tx) error { + close(entered) <-release return nil }) }() - // Let the slow job reach the writer before queuing behind it. - time.Sleep(100 * time.Millisecond) - const waiters = 8 - got := make(chan error, waiters) - for i := 0; i < waiters; i++ { - go func(i int) { - _, err := s.GetOrAssignPseudonym(context.Background(), "ip", fmt.Sprintf("10.9.9.%d", i)) - got <- err - }(i) + select { + case <-entered: + case <-ctx.Done(): + t.Fatal("writer never started the first job") + } + + // Admit jobs explicitly so they are all pending when shutdown starts. + pending := make([]chan error, 8) + for i := range pending { + pending[i] = make(chan error, 1) + if err := s.submit(writeJob{ctx: ctx, done: pending[i], fn: func(*sql.Tx) error { + return fmt.Errorf("queued write ran during shutdown") + }}); err != nil { + t.Fatal(err) + } } - // All waiters must be queued (not saturated) before Close drains. - time.Sleep(200 * time.Millisecond) - // Close waits for the in-flight slow job by design (graceful - // shutdown), so engage it in the background and release the job - // concurrently — serial order would deadlock the test itself. closed := make(chan error, 1) go func() { closed <- s.Close() }() - time.Sleep(100 * time.Millisecond) - close(release) + select { + case <-s.quit: + case <-ctx.Done(): + t.Fatal("close never stopped admission") + } + if err := s.write(ctx, func(*sql.Tx) error { return nil }); !errors.Is(err, ErrStoreClosed) { + t.Fatalf("new write during shutdown: %v", err) + } + unblock() select { case err := <-closed: if err != nil { t.Fatal(err) } - case <-time.After(10 * time.Second): - t.Fatal("Close hung with a drained queue") + case <-ctx.Done(): + t.Fatal("close did not join the writer") } - // The in-flight write either committed before quit won the race - // (nil) or lost it (closed): both are honest outcomes, since the - // caller treats a close as "unknown, retry safely". - if err := <-slowDone; err != nil && !errors.Is(err, ErrStoreClosed) { + if err := <-slowDone; err != nil { t.Fatalf("in-flight write: %v", err) } - // Every queued waiter must RETURN — hanging is the H1 bug. A waiter - // the writer picked before noticing quit ran normally (nil); the - // rest were answered by the drain (closed). - for i := 0; i < waiters; i++ { + if err := s.write(ctx, func(*sql.Tx) error { return nil }); !errors.Is(err, ErrStoreClosed) { + t.Fatalf("new write after shutdown: %v", err) + } + for _, done := range pending { select { - case err := <-got: - if err != nil && !errors.Is(err, ErrStoreClosed) { + case err := <-done: + if !errors.Is(err, ErrStoreClosed) { t.Fatalf("queued write: %v", err) } - case <-time.After(10 * time.Second): - t.Fatal("queued write hung across Close") + default: + t.Fatal("writer exited without rejecting a queued job") } } - // A submit after Close must fail fast, never hang or panic. The - // read path hits the closed pool first, so any prompt error - // satisfies the contract — in-flight calls are the ones owed - // ErrStoreClosed, and they were answered above. - if _, err := s.GetOrAssignPseudonym(context.Background(), "ip", "10.9.9.99"); err == nil { - t.Fatal("post-close submit must fail fast") + if depth := s.QueueDepth(); depth != 0 { + t.Fatalf("closed store retained %d queued jobs", depth) + } +} + +func TestSubmitRacesClose(t *testing.T) { + s, err := Open(filepath.Join(t.TempDir(), "bridge.db"), WithQueueCap(4)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + started := make(chan struct{}) + signalStarted := sync.OnceFunc(func() { close(started) }) + write := func(*sql.Tx) error { return nil } + const writerCount = 32 + var writers sync.WaitGroup + var closedSeen atomic.Int32 + for range writerCount { + writers.Go(func() { + for { + err := s.write(ctx, write) + signalStarted() + switch { + case errors.Is(err, ErrStoreClosed): + closedSeen.Add(1) + return + case err == nil, errors.Is(err, ErrQueueSaturated): + default: + t.Errorf("write: %v", err) + return + } + } + }) + } + <-started + if err := s.Close(); err != nil { + t.Error(err) + } + writers.Wait() + if got := closedSeen.Load(); got != writerCount { + t.Fatalf("%d writers observed closure, want %d", got, writerCount) + } + if depth := s.QueueDepth(); depth != 0 { + t.Fatalf("closed store retained %d queued jobs", depth) } } diff --git a/internal/viewer/viewer.go b/internal/viewer/viewer.go index 92123d5..e8fa371 100644 --- a/internal/viewer/viewer.go +++ b/internal/viewer/viewer.go @@ -107,13 +107,13 @@ func (h *Hub) TrySubscribe(buf, max int) (*Subscription, func(), bool) { if buf <= 0 { buf = 32 } - ch := make(chan Event, buf) - sub := &Subscription{C: ch, ch: ch} h.mu.Lock() defer h.mu.Unlock() if max > 0 && len(h.subs) >= max { return nil, nil, false } + ch := make(chan Event, buf) + sub := &Subscription{C: ch, ch: ch} h.subs[sub] = struct{}{} return sub, func() { h.mu.Lock()