Skip to content

Commit ef4ea49

Browse files
Hyeoncheol Kimclaude
Hyeoncheol Kim
and
claude
committed
Drain async webhook deliveries before storage is closed
Async deliveries ran on a detached context with no way to wait for them, so a delivery could still be writing to SQLite after its caller closed the store. Tests hit this as an intermittent "TempDir RemoveAll cleanup: directory not empty" failure — it turned main's CI red on a test whose whole point is that a slow delivery outlives the request. Service now tracks in-flight deliveries and exposes WaitForAsyncDeliveries, Server.Close drains every run's deliveries before releasing run storage, and test harnesses wait before closing their store. The wait helper lives in a separate webhookstest package so the production binary no longer links testing. Async delivery still returns without blocking its request. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 5c2e18d commit ef4ea49

8 files changed

Lines changed: 177 additions & 89 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,13 @@
22

33
## Unreleased
44

5+
- Asynchronous webhook deliveries are now waitable. They ran on a detached
6+
context with no lifecycle handle, so they could still be writing delivery
7+
records after a caller tore down its storage — which surfaced as intermittent
8+
`TempDir ... directory not empty` test failures. `WaitForAsyncDeliveries`
9+
drains in-flight deliveries, the server drains every run's deliveries on
10+
`Close` before releasing run storage, and tests wait before closing their
11+
store. Async delivery still never blocks the request that triggered it.
512
- Subscription item IDs are now stored when the item is created instead of
613
being derived from its array position, so deleting an item no longer shifts
714
the IDs of the items after it. The `si_<subscription>_<n>` shape is

‎internal/api/api_test.go‎

Lines changed: 58 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
"github.com/hckim/billtap/internal/security"
2424
"github.com/hckim/billtap/internal/storage"
2525
"github.com/hckim/billtap/internal/webhooks"
26+
"github.com/hckim/billtap/internal/webhooks/webhookstest"
2627
)
2728

2829
func TestCheckoutMVPFlow(t *testing.T) {
@@ -91,17 +92,17 @@ func TestCheckoutSessionMetadata(t *testing.T) {
9192

9293
t.Run("payment mode session metadata independent of payment_intent_data", func(t *testing.T) {
9394
session := postForm[map[string]any](t, handler, "/v1/checkout/sessions", url.Values{
94-
"customer": {customer.ID},
95-
"mode": {"payment"},
96-
"client_reference_id": {"ref_meta_1"},
97-
"line_items[0][price_data][currency]": {"usd"},
98-
"line_items[0][price_data][unit_amount]": {"1500"},
99-
"line_items[0][price_data][product_data][name]": {"Extra Export"},
100-
"line_items[0][quantity]": {"1"},
101-
"metadata[paymentType]": {"EXTRA_EXPORT"},
102-
"metadata[accountId]": {"acc_1"},
103-
"payment_intent_data[metadata][paymentType]": {"PI_SIDE"},
104-
"success_url": {"http://app.test/success"},
95+
"customer": {customer.ID},
96+
"mode": {"payment"},
97+
"client_reference_id": {"ref_meta_1"},
98+
"line_items[0][price_data][currency]": {"usd"},
99+
"line_items[0][price_data][unit_amount]": {"1500"},
100+
"line_items[0][price_data][product_data][name]": {"Extra Export"},
101+
"line_items[0][quantity]": {"1"},
102+
"metadata[paymentType]": {"EXTRA_EXPORT"},
103+
"metadata[accountId]": {"acc_1"},
104+
"payment_intent_data[metadata][paymentType]": {"PI_SIDE"},
105+
"success_url": {"http://app.test/success"},
105106
})
106107
meta, _ := session["metadata"].(map[string]any)
107108
if meta["paymentType"] != "EXTRA_EXPORT" || meta["accountId"] != "acc_1" {
@@ -177,12 +178,12 @@ func TestCheckoutSessionMetadata(t *testing.T) {
177178

178179
t.Run("omitted metadata is empty object", func(t *testing.T) {
179180
session := postForm[map[string]any](t, handler, "/v1/checkout/sessions", url.Values{
180-
"customer": {customer.ID},
181-
"mode": {"payment"},
182-
"line_items[0][price_data][currency]": {"usd"},
183-
"line_items[0][price_data][unit_amount]": {"500"},
181+
"customer": {customer.ID},
182+
"mode": {"payment"},
183+
"line_items[0][price_data][currency]": {"usd"},
184+
"line_items[0][price_data][unit_amount]": {"500"},
184185
"line_items[0][price_data][product_data][name]": {"No meta"},
185-
"line_items[0][quantity]": {"1"},
186+
"line_items[0][quantity]": {"1"},
186187
})
187188
meta, ok := session["metadata"].(map[string]any)
188189
if !ok {
@@ -205,12 +206,12 @@ func TestCheckoutSessionMetadata(t *testing.T) {
205206
"provisionAttemptId": "prov_1",
206207
}
207208
values := url.Values{
208-
"customer": {customer.ID},
209-
"mode": {"payment"},
210-
"line_items[0][price_data][currency]": {"usd"},
211-
"line_items[0][price_data][unit_amount]": {"1500"},
209+
"customer": {customer.ID},
210+
"mode": {"payment"},
211+
"line_items[0][price_data][currency]": {"usd"},
212+
"line_items[0][price_data][unit_amount]": {"1500"},
212213
"line_items[0][price_data][product_data][name]": {"Extra Export"},
213-
"line_items[0][quantity]": {"1"},
214+
"line_items[0][quantity]": {"1"},
214215
}
215216
for k, v := range keys {
216217
values.Set("metadata["+k+"]", v)
@@ -244,14 +245,14 @@ func TestCheckoutPaymentMode(t *testing.T) {
244245

245246
t.Run("price_data + payment_intent_data + client_reference_id complete", func(t *testing.T) {
246247
session := postForm[map[string]any](t, handler, "/v1/checkout/sessions", url.Values{
247-
"customer": {customer.ID},
248-
"mode": {"payment"},
249-
"client_reference_id": {"ref_1"},
250-
"line_items[0][price_data][currency]": {"usd"},
251-
"line_items[0][price_data][unit_amount]": {"12900"},
248+
"customer": {customer.ID},
249+
"mode": {"payment"},
250+
"client_reference_id": {"ref_1"},
251+
"line_items[0][price_data][currency]": {"usd"},
252+
"line_items[0][price_data][unit_amount]": {"12900"},
252253
"line_items[0][price_data][product_data][name]": {"Extra export"},
253-
"line_items[0][quantity]": {"1"},
254-
"payment_intent_data[metadata][order_id]": {"ord_1"},
254+
"line_items[0][quantity]": {"1"},
255+
"payment_intent_data[metadata][order_id]": {"ord_1"},
255256
"success_url": {"http://app.test/success"},
256257
"cancel_url": {"http://app.test/cancel"},
257258
})
@@ -320,13 +321,13 @@ func TestCheckoutPaymentMode(t *testing.T) {
320321

321322
t.Run("setup_future_usage attaches payment method", func(t *testing.T) {
322323
session := postForm[map[string]any](t, handler, "/v1/checkout/sessions", url.Values{
323-
"customer": {customer.ID},
324-
"mode": {"payment"},
325-
"line_items[0][price_data][currency]": {"usd"},
326-
"line_items[0][price_data][unit_amount]": {"500"},
324+
"customer": {customer.ID},
325+
"mode": {"payment"},
326+
"line_items[0][price_data][currency]": {"usd"},
327+
"line_items[0][price_data][unit_amount]": {"500"},
327328
"line_items[0][price_data][product_data][name]": {"Attach me"},
328-
"line_items[0][quantity]": {"1"},
329-
"payment_intent_data[setup_future_usage]": {"off_session"},
329+
"line_items[0][quantity]": {"1"},
330+
"payment_intent_data[setup_future_usage]": {"off_session"},
330331
})
331332
completion := postJSON[map[string]json.RawMessage](t, handler, "/api/checkout/sessions/"+fmt.Sprint(session["id"])+"/complete", map[string]string{
332333
// pm_card_visa is a success alias that also stamps a payment_method id.
@@ -412,10 +413,10 @@ func TestCheckoutPaymentMode(t *testing.T) {
412413
"duration": {"once"},
413414
})
414415
session := postForm[map[string]any](t, handler, "/v1/checkout/sessions", url.Values{
415-
"customer": {customer.ID},
416-
"mode": {"payment"},
417-
"line_items[0][price]": {price.ID},
418-
"discounts[0][coupon]": {coupon.ID},
416+
"customer": {customer.ID},
417+
"mode": {"payment"},
418+
"line_items[0][price]": {price.ID},
419+
"discounts[0][coupon]": {coupon.ID},
419420
})
420421
if amount, _ := session["amount_total"].(float64); int64(amount) != 5000 {
421422
t.Fatalf("amount_total=%v, want 5000 after 50%% off", session["amount_total"])
@@ -442,9 +443,9 @@ func TestCheckoutPaymentMode(t *testing.T) {
442443
})
443444

444445
status, body := postFormStatus(t, handler, "/v1/checkout/sessions", url.Values{
445-
"customer": {customer.ID},
446-
"mode": {"subscription"},
447-
"line_items[0][price]": {price.ID},
446+
"customer": {customer.ID},
447+
"mode": {"subscription"},
448+
"line_items[0][price]": {price.ID},
448449
"payment_intent_data[setup_future_usage]": {"off_session"},
449450
})
450451
errBody := decodeErrorBody(t, body)
@@ -464,10 +465,10 @@ func TestCheckoutPaymentMode(t *testing.T) {
464465
}
465466

466467
status, body = postFormStatus(t, handler, "/v1/checkout/sessions", url.Values{
467-
"customer": {customer.ID},
468-
"mode": {"payment"},
469-
"line_items[0][price_data][currency]": {"usd"},
470-
"line_items[0][price_data][unit_amount]": {"100"},
468+
"customer": {customer.ID},
469+
"mode": {"payment"},
470+
"line_items[0][price_data][currency]": {"usd"},
471+
"line_items[0][price_data][unit_amount]": {"100"},
471472
"line_items[0][price_data][product_data][name]": {"Recurring no"},
472473
"line_items[0][price_data][recurring][interval]": {"month"},
473474
})
@@ -505,12 +506,12 @@ func TestCheckoutPaymentMode(t *testing.T) {
505506

506507
t.Run("free payment session no_payment_required", func(t *testing.T) {
507508
session := postForm[map[string]any](t, handler, "/v1/checkout/sessions", url.Values{
508-
"customer": {customer.ID},
509-
"mode": {"payment"},
510-
"line_items[0][price_data][currency]": {"usd"},
511-
"line_items[0][price_data][unit_amount]": {"0"},
509+
"customer": {customer.ID},
510+
"mode": {"payment"},
511+
"line_items[0][price_data][currency]": {"usd"},
512+
"line_items[0][price_data][unit_amount]": {"0"},
512513
"line_items[0][price_data][product_data][name]": {"Freebie"},
513-
"line_items[0][quantity]": {"1"},
514+
"line_items[0][quantity]": {"1"},
514515
})
515516
completion := postJSON[map[string]json.RawMessage](t, handler, "/api/checkout/sessions/"+fmt.Sprint(session["id"])+"/complete", map[string]string{
516517
"outcome": "payment_succeeded",
@@ -6158,15 +6159,12 @@ func TestFixtureApplyBackfillsCreatedEventForExistingSubscriptionSeed(t *testing
61586159
if err != nil {
61596160
t.Fatalf("open sqlite: %v", err)
61606161
}
6161-
t.Cleanup(func() {
6162-
if err := store.Close(); err != nil {
6163-
t.Fatalf("close store: %v", err)
6164-
}
6165-
})
61666162
billingService := billing.NewService(store)
6163+
webhookService := webhooks.NewService(store)
6164+
webhookstest.RegisterStoreCleanup(t, webhookService, store)
61676165
handler := New(Options{
61686166
Billing: billingService,
6169-
Webhooks: webhooks.NewService(store),
6167+
Webhooks: webhookService,
61706168
Diagnostics: diagnostics.NewService(store),
61716169
})
61726170

@@ -6773,14 +6771,12 @@ func newTestHandlerWithOptions(t *testing.T, opts Options) http.Handler {
67736771
if err != nil {
67746772
t.Fatalf("open sqlite: %v", err)
67756773
}
6776-
t.Cleanup(func() {
6777-
if err := store.Close(); err != nil {
6778-
t.Fatalf("close store: %v", err)
6779-
}
6780-
})
67816774
opts.Billing = billing.NewService(store)
6782-
opts.Webhooks = webhooks.NewService(store)
6775+
webhookService := webhooks.NewService(store)
6776+
opts.Webhooks = webhookService
67836777
opts.Diagnostics = diagnostics.NewService(store)
6778+
// After t.TempDir(): LIFO runs wait+close before TempDir removal.
6779+
webhookstest.RegisterStoreCleanup(t, webhookService, store)
67846780
return New(opts)
67856781
}
67866782

‎internal/api/subscription_item_stable_ids_test.go‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import (
1313
"github.com/hckim/billtap/internal/diagnostics"
1414
"github.com/hckim/billtap/internal/storage"
1515
"github.com/hckim/billtap/internal/webhooks"
16+
"github.com/hckim/billtap/internal/webhooks/webhookstest"
1617
)
1718

1819
// newTestHandlerWithStore returns handler + raw store for legacy row injection.
@@ -22,14 +23,12 @@ func newTestHandlerWithStore(t *testing.T) (http.Handler, *storage.SQLiteStore)
2223
if err != nil {
2324
t.Fatalf("open sqlite: %v", err)
2425
}
25-
t.Cleanup(func() {
26-
if err := store.Close(); err != nil {
27-
t.Fatalf("close store: %v", err)
28-
}
29-
})
26+
webhookService := webhooks.NewService(store)
27+
// After t.TempDir(): LIFO runs wait+close before TempDir removal.
28+
webhookstest.RegisterStoreCleanup(t, webhookService, store)
3029
opts := Options{
3130
Billing: billing.NewService(store),
32-
Webhooks: webhooks.NewService(store),
31+
Webhooks: webhookService,
3332
Diagnostics: diagnostics.NewService(store),
3433
}
3534
return New(opts), store

‎internal/server/server.go‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@ import (
88
"os"
99
"path/filepath"
1010
"strings"
11+
"sync"
12+
"time"
1113

1214
"github.com/hckim/billtap/internal/api"
1315
"github.com/hckim/billtap/internal/billing"
@@ -27,6 +29,11 @@ type Server struct {
2729
store storage.Store
2830
mux *http.ServeMux
2931
runs *runManager
32+
33+
// webhookServices tracks every Service built for a run so Close can drain
34+
// in-flight async deliveries before storage is torn down.
35+
webhookMu sync.Mutex
36+
webhookServices []*webhooks.Service
3037
}
3138

3239
func New(opts Options) *Server {
@@ -41,9 +48,21 @@ func New(opts Options) *Server {
4148
return s
4249
}
4350

44-
// Close releases run storage opened on demand. The default store passed
45-
// via Options is owned by the caller and is not closed here.
51+
// Close drains async webhook deliveries for every run, then releases run storage
52+
// opened on demand. The default store passed via Options is owned by the caller
53+
// and is not closed here — callers should Close after this returns so delivery
54+
// records are flushed first.
4655
func (s *Server) Close() error {
56+
s.webhookMu.Lock()
57+
services := append([]*webhooks.Service(nil), s.webhookServices...)
58+
s.webhookMu.Unlock()
59+
if len(services) > 0 {
60+
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
61+
for _, svc := range services {
62+
_ = svc.WaitForAsyncDeliveries(ctx)
63+
}
64+
cancel()
65+
}
4766
if s.runs == nil {
4867
return nil
4968
}
@@ -210,6 +229,9 @@ func (s *Server) buildAPIHandler(store storage.Store) (http.Handler, error) {
210229
SignatureHeaderName: s.cfg.WebhookSignatureHeader,
211230
APIVersion: s.cfg.WebhookAPIVersion,
212231
})
232+
s.webhookMu.Lock()
233+
s.webhookServices = append(s.webhookServices, webhookService)
234+
s.webhookMu.Unlock()
213235
}
214236
var diagnosticsService *diagnostics.Service
215237
if diagnosticsRepo, ok := store.(diagnostics.Repository); ok {

‎internal/server/server_test.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -325,10 +325,15 @@ func newSQLiteBackedServer(t *testing.T, cfg config.Config) http.Handler {
325325
if err != nil {
326326
t.Fatalf("open sqlite: %v", err)
327327
}
328+
srv := New(Options{Config: cfg, Store: store})
329+
// Close drains async webhook deliveries before the default store is closed.
328330
t.Cleanup(func() {
331+
if err := srv.Close(); err != nil {
332+
t.Fatalf("close server: %v", err)
333+
}
329334
if err := store.Close(); err != nil {
330335
t.Fatalf("close store: %v", err)
331336
}
332337
})
333-
return New(Options{Config: cfg, Store: store})
338+
return srv
334339
}

0 commit comments

Comments
 (0)