clawbot
06be075b70
check / check (push) Waiting to run
The receiver, resubmit and replay handlers now read and check the request and make one call. Engine.StoreEvent stores an event with a pending delivery for each active target and the totals in one transaction, then queues them; Engine.CreateReplay does the same for one replay delivery. The handlers take *delivery.Engine, so the Notifier interface and its bridge are gone, and the engine's notify is no longer exported. Archives and CircuitBreakers stay because the handler tests fake them. The tests that read a queued task moved to the engine's tests; the handler tests use an engine that is never started. Model: opus-5-5
482 lines
12 KiB
Go
482 lines
12 KiB
Go
package delivery_test
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"sneak.berlin/go/webhooker/internal/database"
|
|
"sneak.berlin/go/webhooker/internal/delivery"
|
|
)
|
|
|
|
// These tests pin the rule that decides whether a delivery may be
|
|
// handed back to a worker: the engine re-dispatches only what it does
|
|
// not already own. Age alone is not that rule — a healthy delivery
|
|
// waiting in a 10000-deep channel is old and must not be re-sent. See
|
|
// https://git.eeqj.de/sneak/webhooker/issues/256.
|
|
|
|
// fSweepSetup seeds the main database with the webhook row the sweep
|
|
// enumerates, and returns the setup.
|
|
func fSweepSetup(
|
|
t *testing.T, targetID, name string,
|
|
) iSetup {
|
|
t.Helper()
|
|
|
|
s := newISetup(t)
|
|
|
|
iCreateTarget(t, s.MainDB, targetID,
|
|
s.WebhookID, name,
|
|
database.TargetTypeLog, "", 1,
|
|
)
|
|
|
|
require.NoError(t, s.MainDB.Create(&database.Webhook{
|
|
BaseModel: database.BaseModel{ID: s.WebhookID},
|
|
UserID: uuid.New().String(),
|
|
Name: name,
|
|
}).Error)
|
|
|
|
return s
|
|
}
|
|
|
|
// fDrain collects every task the engine has queued.
|
|
//
|
|
// Every caller drives the dispatch paths synchronously and has already
|
|
// waited for them to return, so anything they queued is in the channel
|
|
// by now, and nothing is waited for. A timer here would race the queued
|
|
// tasks: on a busy host it can be due by the time select looks, and
|
|
// select picks at random among the cases that are ready.
|
|
func fDrain(e *delivery.Engine) []delivery.Task {
|
|
var out []delivery.Task
|
|
|
|
for {
|
|
select {
|
|
case task := <-e.ExportDeliveryCh():
|
|
out = append(out, task)
|
|
case task := <-e.ExportRetryCh():
|
|
out = append(out, task)
|
|
default:
|
|
return out
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestSweepLeavesAQueuedDeliveryAlone is the case the age bound cannot
|
|
// see. The delivery is queued and untouched, so its row is arbitrarily
|
|
// old and still perfectly healthy; only ownership distinguishes it
|
|
// from a stranded one.
|
|
func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
targetID := uuid.New().String()
|
|
s := fSweepSetup(t, targetID, "queued")
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"queued":true}`,
|
|
)
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
rAgePending(t, s.WebhookDB, d.ID)
|
|
|
|
// Queued exactly as StoreEvent queues it, and never dequeued:
|
|
// no workers are running in this engine.
|
|
s.Engine.ExportNotify([]delivery.Task{{
|
|
DeliveryID: d.ID,
|
|
EventID: event.ID,
|
|
WebhookID: s.WebhookID,
|
|
TargetID: targetID,
|
|
}})
|
|
|
|
require.Equal(t, 1, s.Engine.ExportInflightHeld())
|
|
|
|
s.Engine.ExportRecoverWebhookDeliveries(
|
|
context.Background(), s.WebhookID, onTick,
|
|
)
|
|
|
|
tasks := fDrain(s.Engine)
|
|
assert.Len(
|
|
t, tasks, 1,
|
|
"the sweep must not queue a delivery that is "+
|
|
"already waiting for a worker",
|
|
)
|
|
}
|
|
|
|
// TestRecoveryAndSweepDoNotDoubleDispatch drives the recovery routine's
|
|
// run at start and a tick of the sweep at once against one aged pending
|
|
// row. The engine runs them one after the other, but notify can race
|
|
// either, and ownership is what keeps any two claimants from both
|
|
// sending. Before ownership the two runs both dispatched the row.
|
|
func TestRecoveryAndSweepDoNotDoubleDispatch(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
targetID := uuid.New().String()
|
|
s := fSweepSetup(t, targetID, "racing")
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"racing":true}`,
|
|
)
|
|
|
|
ctx := context.Background()
|
|
|
|
for range 40 {
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
rAgePending(t, s.WebhookDB, d.ID)
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
wg.Go(func() {
|
|
s.Engine.ExportRecoverPendingDeliveries(
|
|
ctx, s.WebhookDB, s.WebhookID, atStart,
|
|
)
|
|
})
|
|
wg.Go(func() {
|
|
s.Engine.ExportRecoverWebhookDeliveries(
|
|
ctx, s.WebhookID, onTick,
|
|
)
|
|
})
|
|
wg.Wait()
|
|
|
|
tasks := fDrain(s.Engine)
|
|
require.Len(
|
|
t, tasks, 1,
|
|
"delivery %s dispatched %d times",
|
|
d.ID, len(tasks),
|
|
)
|
|
|
|
// No worker runs in this engine, so the reference the winner
|
|
// took is never released and earlier iterations' deliveries
|
|
// stay owned — which is itself the property under test, since
|
|
// both paths see them on every subsequent pass.
|
|
}
|
|
}
|
|
|
|
// TestConcurrentClaimsOfOneDeliveryYieldOneOwner exercises the
|
|
// exclusion directly, rather than arguing it from a SQL predicate.
|
|
func TestConcurrentClaimsOfOneDeliveryYieldOneOwner(
|
|
t *testing.T,
|
|
) {
|
|
t.Parallel()
|
|
|
|
eng := newISetup(t).Engine
|
|
deliveryID := uuid.New().String()
|
|
|
|
var (
|
|
wg sync.WaitGroup
|
|
mu sync.Mutex
|
|
won int
|
|
)
|
|
|
|
for range 64 {
|
|
wg.Go(func() {
|
|
if eng.ExportRetainDelivery(deliveryID) {
|
|
mu.Lock()
|
|
won++
|
|
mu.Unlock()
|
|
}
|
|
})
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
assert.Equal(t, 1, won)
|
|
assert.Equal(t, 1, eng.ExportInflightHeld())
|
|
}
|
|
|
|
// TestOwnershipIsReleasedAfterDelivery guards the other direction: a
|
|
// leaked reference hides a delivery from every sweep for the life of
|
|
// the process.
|
|
func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
s := newISetup(t)
|
|
targetID := uuid.New().String()
|
|
|
|
iCreateTarget(t, s.MainDB, targetID,
|
|
s.WebhookID, "released",
|
|
database.TargetTypeLog, "", 1,
|
|
)
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"released":true}`,
|
|
)
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
s.Engine.ExportStart()
|
|
|
|
defer func() {
|
|
require.NoError(
|
|
t, s.Engine.ExportStop(context.Background()),
|
|
)
|
|
}()
|
|
|
|
body := `{"released":true}`
|
|
|
|
s.Engine.ExportNotify([]delivery.Task{{
|
|
DeliveryID: d.ID,
|
|
EventID: event.ID,
|
|
WebhookID: s.WebhookID,
|
|
TargetID: targetID,
|
|
TargetName: "released",
|
|
TargetType: database.TargetTypeLog,
|
|
Body: &body,
|
|
EntrypointID: event.EntrypointID,
|
|
}})
|
|
|
|
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
|
|
|
assert.Eventually(
|
|
t,
|
|
func() bool {
|
|
return s.Engine.ExportInflightHeld() == 0
|
|
},
|
|
2*time.Second, 20*time.Millisecond,
|
|
"the delivery stayed owned after it was delivered",
|
|
)
|
|
}
|
|
|
|
// TestNotifyAfterRecoveryDoesNotSendAgain is the startup race of
|
|
// https://git.eeqj.de/sneak/webhooker/issues/299. StoreEvent has
|
|
// written a delivery, the recovery routine's run at start finds it
|
|
// pending, sends it and releases it, and only then does StoreEvent's
|
|
// notify for it arrive. Nothing owns the delivery by then, so notify
|
|
// takes it.
|
|
func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
targetID := uuid.New().String()
|
|
s := fSweepSetup(t, targetID, "recovered")
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"recovered":true}`,
|
|
)
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
s.Engine.ExportStart()
|
|
|
|
defer func() {
|
|
require.NoError(
|
|
t, s.Engine.ExportStop(context.Background()),
|
|
)
|
|
}()
|
|
|
|
// The recovery routine's run at start sends the delivery and lets
|
|
// it go.
|
|
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
|
require.Eventually(
|
|
t,
|
|
func() bool {
|
|
return s.Engine.ExportInflightHeld() == 0
|
|
},
|
|
5*time.Second, 20*time.Millisecond,
|
|
)
|
|
|
|
body := event.Body
|
|
|
|
s.Engine.ExportNotify([]delivery.Task{{
|
|
DeliveryID: d.ID,
|
|
EventID: event.ID,
|
|
WebhookID: s.WebhookID,
|
|
TargetID: targetID,
|
|
TargetName: "recovered",
|
|
TargetType: database.TargetTypeLog,
|
|
Body: &body,
|
|
EntrypointID: event.EntrypointID,
|
|
}})
|
|
|
|
// notify took the delivery, and a worker releases it once it has
|
|
// run the task.
|
|
require.Eventually(
|
|
t,
|
|
func() bool {
|
|
return s.Engine.ExportInflightHeld() == 0
|
|
},
|
|
5*time.Second, 20*time.Millisecond,
|
|
)
|
|
|
|
assert.Len(
|
|
t, iResults(t, s.WebhookDB, d.ID), 1,
|
|
"the delivery was sent a second time",
|
|
)
|
|
}
|
|
|
|
// TestRetryingRecoverySkipsASuccessfulResult is the retrying-side twin
|
|
// of the pending reconcile. A second attempt that reached the receiver
|
|
// and whose status write then failed sits at retrying holding a
|
|
// successful result, and re-sending it is the same duplicate.
|
|
func TestRetryingRecoverySkipsASuccessfulResult(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
targetID := uuid.New().String()
|
|
s := fSweepSetup(t, targetID, "retry-settled")
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"retry":true}`,
|
|
)
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusRetrying,
|
|
)
|
|
rSeedResult(t, s.WebhookDB, d.ID, 1, false)
|
|
rSeedResult(t, s.WebhookDB, d.ID, 2, true)
|
|
|
|
s.Engine.ExportRecoverRetryingDeliveries(
|
|
context.Background(), s.WebhookDB, s.WebhookID,
|
|
)
|
|
|
|
assert.Empty(
|
|
t, fDrain(s.Engine),
|
|
"a retrying delivery holding a successful result "+
|
|
"must not be sent again",
|
|
)
|
|
|
|
iAssertStatus(
|
|
t, s.WebhookDB, d.ID,
|
|
database.DeliveryStatusDelivered,
|
|
)
|
|
}
|
|
|
|
// TestRetryingSweepSkipsASuccessfulResult is the same rule on a tick of
|
|
// the sweep.
|
|
func TestRetryingSweepSkipsASuccessfulResult(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
targetID := uuid.New().String()
|
|
s := fSweepSetup(t, targetID, "retry-swept")
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"swept":true}`,
|
|
)
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusRetrying,
|
|
)
|
|
rSeedResult(t, s.WebhookDB, d.ID, 1, false)
|
|
rSeedResult(t, s.WebhookDB, d.ID, 2, true)
|
|
|
|
s.Engine.ExportRecoverWebhookDeliveries(
|
|
context.Background(), s.WebhookID, onTick,
|
|
)
|
|
|
|
assert.Empty(t, fDrain(s.Engine))
|
|
|
|
iAssertStatus(
|
|
t, s.WebhookDB, d.ID,
|
|
database.DeliveryStatusDelivered,
|
|
)
|
|
|
|
var attempts int64
|
|
|
|
require.NoError(t, s.WebhookDB.
|
|
Model(&database.DeliveryResult{}).
|
|
Where("delivery_id = ?", d.ID).
|
|
Count(&attempts).Error)
|
|
assert.Equal(
|
|
t, int64(2), attempts,
|
|
"settling must not invent an attempt",
|
|
)
|
|
}
|
|
|
|
// TestScheduledRetryIsNotSweptDuringBackoff closes the window between
|
|
// a target scheduling a retry and the timer firing. The row says
|
|
// retrying and nothing is running, which is exactly what an orphaned
|
|
// retry looks like from the database.
|
|
func TestScheduledRetryIsNotSweptDuringBackoff(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
targetID := uuid.New().String()
|
|
s := fSweepSetup(t, targetID, "backoff")
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"backoff":true}`,
|
|
)
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusRetrying,
|
|
)
|
|
|
|
s.Engine.ExportScheduleRetry(delivery.Task{
|
|
DeliveryID: d.ID,
|
|
EventID: event.ID,
|
|
WebhookID: s.WebhookID,
|
|
TargetID: targetID,
|
|
AttemptNum: 2,
|
|
}, time.Hour)
|
|
|
|
require.Equal(t, 1, s.Engine.ExportInflightHeld())
|
|
|
|
s.Engine.ExportRecoverWebhookDeliveries(
|
|
context.Background(), s.WebhookID, onTick,
|
|
)
|
|
|
|
assert.Empty(
|
|
t, fDrain(s.Engine),
|
|
"the sweep must not duplicate a retry that is "+
|
|
"already scheduled",
|
|
)
|
|
}
|
|
|
|
// TestRedispatchStampsTheRow pins the cadence control: a stranded
|
|
// delivery that has just been handed out is not selected again by the
|
|
// next tick a minute later. The stamp is stored in the same form as
|
|
// created_at.
|
|
func TestRedispatchStampsTheRow(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
targetID := uuid.New().String()
|
|
s := fSweepSetup(t, targetID, "stamped")
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"stamped":true}`,
|
|
)
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
rAgePending(t, s.WebhookDB, d.ID)
|
|
|
|
ctx := context.Background()
|
|
|
|
s.Engine.ExportRecoverWebhookDeliveries(ctx, s.WebhookID, onTick)
|
|
require.Len(t, fDrain(s.Engine), 1)
|
|
|
|
var row database.Delivery
|
|
|
|
require.NoError(t, s.WebhookDB.
|
|
First(&row, "id = ?", d.ID).Error)
|
|
assert.WithinDuration(
|
|
t, time.Now(), row.UpdatedAt, time.Minute,
|
|
"a re-dispatched delivery must be stamped so the "+
|
|
"next tick does not select it again",
|
|
)
|
|
|
|
var createdAt, updatedAt string
|
|
|
|
require.NoError(t, s.WebhookDB.Raw(
|
|
`SELECT CAST(created_at AS TEXT), CAST(updated_at AS TEXT)
|
|
FROM deliveries WHERE id = ?`, d.ID,
|
|
).Row().Scan(&createdAt, &updatedAt))
|
|
assertStoredTime(t, createdAt)
|
|
assertStoredTime(t, updatedAt)
|
|
}
|