Skip to content

Commit baa83e6

Browse files
mekilismekilis
authored
fix(dataplane): capture delivery attempt timestamps around the dispatch call (#2707)
requested_at was stamped at httpDispatchStart, before signature/header/mTLS prep, and responded_at was httpDispatchStart + elapsed duration, so responded_at - requested_at overstated real endpoint latency by our own prep time. Capture requestSentAt immediately before SendWebhookWithMTLS and responseReceivedAt immediately after it in both the primary and retry paths, and pass those to parseAttemptFromResponse. The resp != nil && StatusCode >= 100 guard is unchanged, so responded_at stays null on no-response. Adds flow tests driving a slow loopback server through a real dispatcher to assert responded_at - requested_at reflects the actual round trip.
1 parent bc0b5ec commit baa83e6

4 files changed

Lines changed: 253 additions & 4 deletions

File tree

‎worker/task/process_event_delivery.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -355,7 +355,9 @@ func ProcessEventDelivery(deps EventDeliveryProcessorDeps) func(context.Context,
355355
}
356356
}
357357

358+
requestSentAt := time.Now()
358359
resp, err := deps.Dispatcher.SendWebhookWithMTLS(ctx, targetURL, sig.Payload, project.Config.Signature.Header.String(), header, int64(cfg.MaxResponseSize), eventDelivery.Headers, eventDelivery.IdempotencyKey, httpDuration, contentType, mtlsCert)
360+
responseReceivedAt := time.Now()
359361

360362
status := "-"
361363
statusCode := 0
@@ -436,9 +438,9 @@ func ProcessEventDelivery(deps EventDeliveryProcessorDeps) func(context.Context,
436438

437439
respondedAt := time.Time{}
438440
if resp != nil && resp.StatusCode >= 100 {
439-
respondedAt = httpDispatchStart.Add(duration)
441+
respondedAt = responseReceivedAt
440442
}
441-
attempt := parseAttemptFromResponse(eventDelivery, endpoint, resp, attemptStatus, httpDispatchStart, respondedAt)
443+
attempt := parseAttemptFromResponse(eventDelivery, endpoint, resp, attemptStatus, requestSentAt, respondedAt)
442444
eventDelivery.Metadata.NumTrials++
443445

444446
if eventDelivery.Metadata.NumTrials >= eventDelivery.Metadata.RetryLimit {

‎worker/task/process_event_delivery_test.go‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1404,3 +1404,127 @@ func TestProcessEventDeliveryConfig(t *testing.T) {
14041404
})
14051405
}
14061406
}
1407+
1408+
// TestProcessEventDelivery_AttemptTimingReflectsWireDelay proves the attempt's
1409+
// requested_at/responded_at bracket the actual dispatch call: the gap tracks the
1410+
// real round trip instead of collapsing to ~0 or including our pre-send prep time.
1411+
func TestProcessEventDelivery_AttemptTimingReflectsWireDelay(t *testing.T) {
1412+
const wireDelay = 60 * time.Millisecond
1413+
slowServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
1414+
time.Sleep(wireDelay)
1415+
w.WriteHeader(http.StatusBadRequest)
1416+
}))
1417+
defer slowServer.Close()
1418+
1419+
ctrl := gomock.NewController(t)
1420+
defer ctrl.Finish()
1421+
1422+
endpointRepo := mocks.NewMockEndpointRepository(ctrl)
1423+
projectRepo := mocks.NewMockProjectRepository(ctrl)
1424+
msgRepo := mocks.NewMockEventDeliveryRepository(ctrl)
1425+
q := mocks.NewMockQueuer(ctrl)
1426+
rateLimiter := mocks.NewMockRateLimiter(ctrl)
1427+
attemptsRepo := mocks.NewMockDeliveryAttemptsRepository(ctrl)
1428+
licenser := mocks.NewMockLicenser(ctrl)
1429+
licenser.EXPECT().ProjectEnabled(gomock.Any()).Return(true).AnyTimes()
1430+
licenser.EXPECT().UseForwardProxy().Return(true).AnyTimes()
1431+
licenser.EXPECT().IpRules().Return(true).AnyTimes()
1432+
licenser.EXPECT().AdvancedEndpointMgmt().Return(false).AnyTimes()
1433+
licenser.EXPECT().CircuitBreaking().Return(false).AnyTimes()
1434+
1435+
require.NoError(t, config.LoadConfig("./testdata/Config/basic-convoy.json"))
1436+
1437+
endpointRepo.EXPECT().FindEndpointByID(gomock.Any(), gomock.Any(), gomock.Any()).
1438+
Return(&datastore.Endpoint{
1439+
UID: "endpoint-id-1",
1440+
ProjectID: "123",
1441+
Url: slowServer.URL,
1442+
RateLimit: 10,
1443+
RateLimitDuration: 60,
1444+
Secrets: []datastore.Secret{{Value: "secret"}},
1445+
Status: datastore.ActiveEndpointStatus,
1446+
}, nil)
1447+
1448+
rateLimiter.EXPECT().AllowWithDuration(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
1449+
1450+
// At-most-once with a non-2xx response fails without a retry, so the flow
1451+
// returns nil with no retry enqueue: the minimal path that still records an
1452+
// attempt with both timestamps.
1453+
msgRepo.EXPECT().FindEventDeliveryByIDSlim(gomock.Any(), gomock.Any(), gomock.Any()).
1454+
Return(&datastore.EventDelivery{
1455+
UID: "evt-del-1",
1456+
EndpointID: "endpoint-id-1",
1457+
SubscriptionID: "sub-id-1",
1458+
ProjectID: "123",
1459+
Metadata: &datastore.Metadata{
1460+
Data: []byte(`{"event":"x"}`),
1461+
Raw: `{"event":"x"}`,
1462+
NumTrials: 0,
1463+
RetryLimit: 3,
1464+
IntervalSeconds: 20,
1465+
},
1466+
Status: datastore.ScheduledEventStatus,
1467+
DeliveryMode: datastore.AtMostOnceDeliveryMode,
1468+
}, nil)
1469+
1470+
projectRepo.EXPECT().FetchProjectByID(gomock.Any(), gomock.Any()).
1471+
Return(&datastore.Project{
1472+
UID: "123",
1473+
Config: &datastore.ProjectConfig{
1474+
Signature: &datastore.SignatureConfiguration{
1475+
Header: "X-Convoy-Signature",
1476+
Versions: []datastore.SignatureVersion{{UID: "abc", Hash: "SHA256", Encoding: datastore.HexEncoding}},
1477+
},
1478+
SSL: &datastore.DefaultSSLConfig,
1479+
Strategy: &datastore.StrategyConfiguration{Type: datastore.LinearStrategyProvider, Duration: 60, RetryCount: 3},
1480+
RateLimit: &datastore.DefaultRateLimitConfig,
1481+
},
1482+
}, nil)
1483+
1484+
msgRepo.EXPECT().UpdateStatusOfEventDelivery(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(1)
1485+
1486+
var captured *datastore.DeliveryAttempt
1487+
attemptsRepo.EXPECT().CreateDeliveryAttempt(gomock.Any(), gomock.Any()).
1488+
DoAndReturn(func(_ context.Context, a *datastore.DeliveryAttempt) error {
1489+
captured = a
1490+
return nil
1491+
}).Times(1)
1492+
1493+
msgRepo.EXPECT().UpdateEventDeliveryMetadata(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(1)
1494+
1495+
// Plain dispatcher: no IpRules flag, blocklist, or proxy, so the delivery
1496+
// reaches the loopback test server for a real, measurable round trip.
1497+
dispatcher, err := net.NewDispatcher(
1498+
licenser,
1499+
fflag.NewFFlag([]string{}),
1500+
net.LoggerOption(log.New("convoy", log.LevelInfo)),
1501+
)
1502+
require.NoError(t, err)
1503+
1504+
deps := EventDeliveryProcessorDeps{
1505+
EndpointRepo: endpointRepo,
1506+
EventDeliveryRepo: msgRepo,
1507+
Licenser: licenser,
1508+
ProjectRepo: projectRepo,
1509+
Queue: q,
1510+
RateLimiter: rateLimiter,
1511+
Dispatcher: dispatcher,
1512+
AttemptsRepo: attemptsRepo,
1513+
FeatureFlag: fflag.NewFFlag([]string{}),
1514+
Logger: log.New("convoy", log.LevelInfo),
1515+
}
1516+
1517+
payload, err := json.Marshal(EventDelivery{EventDeliveryID: "evt-del-1", ProjectID: "123"})
1518+
require.NoError(t, err)
1519+
task := asynq.NewTask(string(convoy.EventProcessor), payload, asynq.Queue(string(convoy.EventQueue)))
1520+
1521+
require.NoError(t, ProcessEventDelivery(deps)(context.Background(), task))
1522+
1523+
require.NotNil(t, captured, "expected a delivery attempt to be created")
1524+
require.True(t, captured.RequestedAt.Valid, "requested_at should be set")
1525+
require.True(t, captured.RespondedAt.Valid, "responded_at should be set on a completed round trip")
1526+
require.False(t, captured.RespondedAt.Time.Before(captured.RequestedAt.Time), "responded_at must not precede requested_at")
1527+
1528+
gap := captured.RespondedAt.Time.Sub(captured.RequestedAt.Time)
1529+
require.GreaterOrEqual(t, gap, wireDelay/2, "responded_at - requested_at should reflect the real round trip, got %s", gap)
1530+
}

‎worker/task/process_retry_event_delivery.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -289,7 +289,9 @@ func ProcessRetryEventDelivery(deps EventDeliveryProcessorDeps) func(context.Con
289289
}
290290
}
291291

292+
requestSentAt := time.Now()
292293
resp, err := deps.Dispatcher.SendWebhookWithMTLS(ctx, targetURL, sig.Payload, project.Config.Signature.Header.String(), header, int64(cfg.MaxResponseSize), eventDelivery.Headers, eventDelivery.IdempotencyKey, httpDuration, contentType, mtlsCert)
294+
responseReceivedAt := time.Now()
293295

294296
status := "-"
295297
statusCode := 0
@@ -359,9 +361,9 @@ func ProcessRetryEventDelivery(deps EventDeliveryProcessorDeps) func(context.Con
359361

360362
respondedAt := time.Time{}
361363
if resp != nil && resp.StatusCode >= 100 {
362-
respondedAt = httpDispatchStart.Add(duration)
364+
respondedAt = responseReceivedAt
363365
}
364-
attempt = parseAttemptFromResponse(eventDelivery, endpoint, resp, attemptStatus, httpDispatchStart, respondedAt)
366+
attempt = parseAttemptFromResponse(eventDelivery, endpoint, resp, attemptStatus, requestSentAt, respondedAt)
365367

366368
eventDelivery.Metadata.NumTrials++
367369

‎worker/task/process_retry_event_delivery_test.go‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1304,3 +1304,124 @@ func TestProcessRetryEventDeliveryConfig(t *testing.T) {
13041304
})
13051305
}
13061306
}
1307+
1308+
// TestProcessRetryEventDelivery_AttemptTimingReflectsWireDelay mirrors the
1309+
// primary-path timing test for the retry worker: the recorded attempt's
1310+
// requested_at/responded_at must bracket the real dispatch call.
1311+
func TestProcessRetryEventDelivery_AttemptTimingReflectsWireDelay(t *testing.T) {
1312+
const wireDelay = 60 * time.Millisecond
1313+
slowServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
1314+
time.Sleep(wireDelay)
1315+
w.WriteHeader(http.StatusBadRequest)
1316+
}))
1317+
defer slowServer.Close()
1318+
1319+
ctrl := gomock.NewController(t)
1320+
defer ctrl.Finish()
1321+
1322+
endpointRepo := mocks.NewMockEndpointRepository(ctrl)
1323+
projectRepo := mocks.NewMockProjectRepository(ctrl)
1324+
msgRepo := mocks.NewMockEventDeliveryRepository(ctrl)
1325+
q := mocks.NewMockQueuer(ctrl)
1326+
rateLimiter := mocks.NewMockRateLimiter(ctrl)
1327+
attemptsRepo := mocks.NewMockDeliveryAttemptsRepository(ctrl)
1328+
licenser := mocks.NewMockLicenser(ctrl)
1329+
licenser.EXPECT().ProjectEnabled(gomock.Any()).Return(true).AnyTimes()
1330+
licenser.EXPECT().UseForwardProxy().Return(true).AnyTimes()
1331+
licenser.EXPECT().IpRules().Return(true).AnyTimes()
1332+
licenser.EXPECT().AdvancedEndpointMgmt().Return(false).AnyTimes()
1333+
licenser.EXPECT().CircuitBreaking().Return(false).AnyTimes()
1334+
1335+
require.NoError(t, config.LoadConfig("./testdata/Config/basic-convoy.json"))
1336+
1337+
endpointRepo.EXPECT().FindEndpointByID(gomock.Any(), gomock.Any(), gomock.Any()).
1338+
Return(&datastore.Endpoint{
1339+
UID: "endpoint-id-1",
1340+
ProjectID: "123",
1341+
Url: slowServer.URL,
1342+
RateLimit: 10,
1343+
RateLimitDuration: 60,
1344+
Secrets: []datastore.Secret{{Value: "secret"}},
1345+
Status: datastore.ActiveEndpointStatus,
1346+
}, nil)
1347+
1348+
rateLimiter.EXPECT().AllowWithDuration(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
1349+
1350+
msgRepo.EXPECT().FindEventDeliveryByID(gomock.Any(), gomock.Any(), gomock.Any()).
1351+
Return(&datastore.EventDelivery{
1352+
UID: "evt-del-1",
1353+
EndpointID: "endpoint-id-1",
1354+
SubscriptionID: "sub-id-1",
1355+
ProjectID: "123",
1356+
Metadata: &datastore.Metadata{
1357+
Data: []byte(`{"event":"x"}`),
1358+
Raw: `{"event":"x"}`,
1359+
NumTrials: 0,
1360+
RetryLimit: 3,
1361+
IntervalSeconds: 20,
1362+
},
1363+
Status: datastore.ScheduledEventStatus,
1364+
DeliveryMode: datastore.AtMostOnceDeliveryMode,
1365+
}, nil)
1366+
1367+
projectRepo.EXPECT().FetchProjectByID(gomock.Any(), gomock.Any()).
1368+
Return(&datastore.Project{
1369+
UID: "123",
1370+
Config: &datastore.ProjectConfig{
1371+
Signature: &datastore.SignatureConfiguration{
1372+
Header: "X-Convoy-Signature",
1373+
Versions: []datastore.SignatureVersion{{UID: "abc", Hash: "SHA256", Encoding: datastore.HexEncoding}},
1374+
},
1375+
SSL: &datastore.DefaultSSLConfig,
1376+
Strategy: &datastore.StrategyConfiguration{Type: datastore.LinearStrategyProvider, Duration: 60, RetryCount: 3},
1377+
RateLimit: &datastore.DefaultRateLimitConfig,
1378+
},
1379+
}, nil)
1380+
1381+
msgRepo.EXPECT().UpdateStatusOfEventDelivery(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(1)
1382+
1383+
var captured *datastore.DeliveryAttempt
1384+
attemptsRepo.EXPECT().CreateDeliveryAttempt(gomock.Any(), gomock.Any()).
1385+
DoAndReturn(func(_ context.Context, a *datastore.DeliveryAttempt) error {
1386+
captured = a
1387+
return nil
1388+
}).Times(1)
1389+
1390+
msgRepo.EXPECT().UpdateEventDeliveryMetadata(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(1)
1391+
1392+
// Plain dispatcher: no IpRules flag, blocklist, or proxy, so the delivery
1393+
// reaches the loopback test server for a real, measurable round trip.
1394+
dispatcher, err := net.NewDispatcher(
1395+
licenser,
1396+
fflag.NewFFlag([]string{}),
1397+
net.LoggerOption(log.New("convoy", log.LevelInfo)),
1398+
)
1399+
require.NoError(t, err)
1400+
1401+
deps := EventDeliveryProcessorDeps{
1402+
EndpointRepo: endpointRepo,
1403+
EventDeliveryRepo: msgRepo,
1404+
Licenser: licenser,
1405+
ProjectRepo: projectRepo,
1406+
Queue: q,
1407+
RateLimiter: rateLimiter,
1408+
Dispatcher: dispatcher,
1409+
AttemptsRepo: attemptsRepo,
1410+
FeatureFlag: fflag.NewFFlag([]string{}),
1411+
Logger: log.New("convoy", log.LevelInfo),
1412+
}
1413+
1414+
payload, err := json.Marshal(EventDelivery{EventDeliveryID: "evt-del-1", ProjectID: "123"})
1415+
require.NoError(t, err)
1416+
task := asynq.NewTask(string(convoy.EventProcessor), payload, asynq.Queue(string(convoy.EventQueue)))
1417+
1418+
require.NoError(t, ProcessRetryEventDelivery(deps)(context.Background(), task))
1419+
1420+
require.NotNil(t, captured, "expected a delivery attempt to be created")
1421+
require.True(t, captured.RequestedAt.Valid, "requested_at should be set")
1422+
require.True(t, captured.RespondedAt.Valid, "responded_at should be set on a completed round trip")
1423+
require.False(t, captured.RespondedAt.Time.Before(captured.RequestedAt.Time), "responded_at must not precede requested_at")
1424+
1425+
gap := captured.RespondedAt.Time.Sub(captured.RequestedAt.Time)
1426+
require.GreaterOrEqual(t, gap, wireDelay/2, "responded_at - requested_at should reflect the real round trip, got %s", gap)
1427+
}

0 commit comments

Comments
 (0)