Use one recovery routine at start and on every sweep tick (closes #542)
check / check (push) Waiting to run
check / check (push) Waiting to run
Start-up recovery and the 60-second sweep were two parallel sets of functions doing the same job. recoverDeliveries now does it, run once at start and then on every tick from one goroutine. A retrying delivery is scheduled after what is left of its backoff, now if nothing is left, so backoffElapsed is gone. The run at start still takes every pending delivery and opens a webhook database whose file is missing; a tick still takes only pending deliveries untouched for 15 minutes, at most 500, and skips a missing file. A retrying delivery is taken before its event is read. The README describes one routine. Model: opus-5-5
This commit is contained in:
@@ -95,8 +95,8 @@ func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) {
|
||||
|
||||
require.Equal(t, 1, s.Engine.ExportInflightHeld())
|
||||
|
||||
s.Engine.ExportSweepWebhookRetries(
|
||||
context.Background(), s.WebhookID,
|
||||
s.Engine.ExportRecoverWebhookDeliveries(
|
||||
context.Background(), s.WebhookID, onTick,
|
||||
)
|
||||
|
||||
tasks := fDrain(s.Engine)
|
||||
@@ -107,9 +107,11 @@ func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) {
|
||||
)
|
||||
}
|
||||
|
||||
// TestRecoveryAndSweepDoNotDoubleDispatch drives the two entry points
|
||||
// the engine starts concurrently against one aged pending row. Before
|
||||
// ownership they both dispatched it.
|
||||
// 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()
|
||||
|
||||
@@ -133,12 +135,12 @@ func TestRecoveryAndSweepDoNotDoubleDispatch(t *testing.T) {
|
||||
|
||||
wg.Go(func() {
|
||||
s.Engine.ExportRecoverPendingDeliveries(
|
||||
ctx, s.WebhookDB, s.WebhookID,
|
||||
ctx, s.WebhookDB, s.WebhookID, atStart,
|
||||
)
|
||||
})
|
||||
wg.Go(func() {
|
||||
s.Engine.ExportSweepWebhookRetries(
|
||||
ctx, s.WebhookID,
|
||||
s.Engine.ExportRecoverWebhookDeliveries(
|
||||
ctx, s.WebhookID, onTick,
|
||||
)
|
||||
})
|
||||
wg.Wait()
|
||||
@@ -247,9 +249,10 @@ func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
|
||||
|
||||
// TestNotifyAfterRecoveryDoesNotSendAgain is the startup race of
|
||||
// https://git.eeqj.de/sneak/webhooker/issues/299. The receiver has
|
||||
// written a delivery, restart recovery finds it pending, sends it and
|
||||
// releases it, and only then does the receiver's Notify for it arrive.
|
||||
// Nothing owns the delivery by then, so Notify takes it.
|
||||
// written a delivery, the recovery routine's run at start finds it
|
||||
// pending, sends it and releases it, and only then does the receiver's
|
||||
// Notify for it arrive. Nothing owns the delivery by then, so Notify
|
||||
// takes it.
|
||||
func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -273,7 +276,8 @@ func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {
|
||||
)
|
||||
}()
|
||||
|
||||
// Restart recovery sends the delivery and lets it go.
|
||||
// The recovery routine's run at start sends the delivery and lets
|
||||
// it go.
|
||||
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
||||
require.Eventually(
|
||||
t,
|
||||
@@ -334,7 +338,7 @@ func TestRetryingRecoverySkipsASuccessfulResult(t *testing.T) {
|
||||
rSeedResult(t, s.WebhookDB, d.ID, 2, true)
|
||||
|
||||
s.Engine.ExportRecoverRetryingDeliveries(
|
||||
s.WebhookDB, s.WebhookID,
|
||||
context.Background(), s.WebhookDB, s.WebhookID,
|
||||
)
|
||||
|
||||
assert.Empty(
|
||||
@@ -349,8 +353,8 @@ func TestRetryingRecoverySkipsASuccessfulResult(t *testing.T) {
|
||||
)
|
||||
}
|
||||
|
||||
// TestRetryingSweepSkipsASuccessfulResult is the same rule on the
|
||||
// periodic sweep's retrying arm.
|
||||
// TestRetryingSweepSkipsASuccessfulResult is the same rule on a tick of
|
||||
// the sweep.
|
||||
func TestRetryingSweepSkipsASuccessfulResult(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -368,8 +372,8 @@ func TestRetryingSweepSkipsASuccessfulResult(t *testing.T) {
|
||||
rSeedResult(t, s.WebhookDB, d.ID, 1, false)
|
||||
rSeedResult(t, s.WebhookDB, d.ID, 2, true)
|
||||
|
||||
s.Engine.ExportSweepWebhookRetries(
|
||||
context.Background(), s.WebhookID,
|
||||
s.Engine.ExportRecoverWebhookDeliveries(
|
||||
context.Background(), s.WebhookID, onTick,
|
||||
)
|
||||
|
||||
assert.Empty(t, fDrain(s.Engine))
|
||||
@@ -420,8 +424,8 @@ func TestScheduledRetryIsNotSweptDuringBackoff(t *testing.T) {
|
||||
|
||||
require.Equal(t, 1, s.Engine.ExportInflightHeld())
|
||||
|
||||
s.Engine.ExportSweepWebhookRetries(
|
||||
context.Background(), s.WebhookID,
|
||||
s.Engine.ExportRecoverWebhookDeliveries(
|
||||
context.Background(), s.WebhookID, onTick,
|
||||
)
|
||||
|
||||
assert.Empty(
|
||||
@@ -453,7 +457,7 @@ func TestRedispatchStampsTheRow(t *testing.T) {
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
s.Engine.ExportSweepWebhookRetries(ctx, s.WebhookID)
|
||||
s.Engine.ExportRecoverWebhookDeliveries(ctx, s.WebhookID, onTick)
|
||||
require.Len(t, fDrain(s.Engine), 1)
|
||||
|
||||
var row database.Delivery
|
||||
|
||||
Reference in New Issue
Block a user