webhooker/internal/delivery/recovery_test.go

package delivery_test

import (
	"context"
	"log/slog"
	"net/http"
	"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/database/databasetest"
	"sneak.berlin/go/webhooker/internal/delivery"
)

// These tests cover the recovery routine, which the engine runs once
// when it starts and then on every tick of the sweep
// (https://git.eeqj.de/sneak/webhooker/issues/542). Every case runs it
// both ways, each on a webhook of its own.

// atStart and onTick are the two values of the recovery routine's
// atStart parameter: its run when the engine starts, and a tick of the
// sweep.
const (
	atStart = true
	onTick  = false
)

// recoveryRun names one of the two runs of the recovery routine.
type recoveryRun struct {
	name    string
	atStart bool
}

// recoveryRuns returns the routine's run at start and a tick.
func recoveryRuns() []recoveryRun {
	return []recoveryRun{
		{"at start", atStart},
		{"on a tick", onTick},
	}
}

// recoverySetup gives a fresh webhook its main-database row and an http
// target with retries, and returns the setup and the target's ID. No
// worker runs, so whatever the routine queues stays in the channels for
// the test to read, and nothing is sent to the target.
func recoverySetup(t *testing.T) (iSetup, string) {
	t.Helper()

	s := newISetup(t)
	targetID := uuid.New().String()

	iCreateWebhook(t, s.MainDB, s.WebhookID, "recovery")
	iCreateTarget(
		t, s.MainDB, targetID, s.WebhookID, "recovery",
		database.TargetTypeHTTP,
		"http://example.com/hook", 20,
	)

	return s, targetID
}

// seedRecoveryDelivery seeds a delivery at status with the given
// recorded attempts, failed but for the last when lastSucceeded is set,
// all recorded now.
func seedRecoveryDelivery(
	t *testing.T,
	s iSetup,
	targetID string,
	status database.DeliveryStatus,
	attempts int,
	lastSucceeded bool,
) string {
	t.Helper()

	event := iSeedEvent(t, s.WebhookDB, s.WebhookID, `{"recovery":true}`)
	d := iSeedDelivery(t, s.WebhookDB, event.ID, targetID, status)

	for n := 1; n <= attempts; n++ {
		rSeedResult(
			t, s.WebhookDB, d.ID, n, lastSucceeded && n == attempts,
		)
	}

	return d.ID
}

// backdateAttempts moves every recorded attempt of a delivery an hour
// into the past, long past the backoff after its first attempt.
func backdateAttempts(t *testing.T, s iSetup, deliveryID string) {
	t.Helper()

	require.NoError(t, s.WebhookDB.Model(&database.DeliveryResult{}).
		Where("delivery_id = ?", deliveryID).
		UpdateColumn("created_at", s.WebhookDB.NowFunc().Add(-time.Hour)).
		Error)
}

// requireQueued waits for a task on ch. A retrying delivery reaches the
// retry channel through a timer, even one due now, so it is waited for.
func requireQueued(
	t *testing.T, ch <-chan delivery.Task,
) delivery.Task {
	t.Helper()

	select {
	case task := <-ch:
		return task
	case <-time.After(5 * time.Second):
		t.Fatal("the recovery routine queued nothing")

		return delivery.Task{}
	}
}

// assertNothingQueued asserts the engine queues nothing, giving a retry
// timer that is due now the moment it needs to fire.
func assertNothingQueued(t *testing.T, e *delivery.Engine) {
	t.Helper()

	select {
	case task := <-e.ExportDeliveryCh():
		t.Errorf("queued pending delivery %s", task.DeliveryID)
	case task := <-e.ExportRetryCh():
		t.Errorf("queued retrying delivery %s", task.DeliveryID)
	case <-time.After(200 * time.Millisecond):
	}
}

// TestRecovery_RetryingBackoffPassed: a retrying delivery whose backoff
// is over is queued for its next attempt at once.
func TestRecovery_RetryingBackoffPassed(t *testing.T) {
	t.Parallel()

	for _, run := range recoveryRuns() {
		t.Run(run.name, func(t *testing.T) {
			t.Parallel()

			s, targetID := recoverySetup(t)
			id := seedRecoveryDelivery(
				t, s, targetID, database.DeliveryStatusRetrying, 1, false,
			)
			backdateAttempts(t, s, id)

			s.Engine.ExportRecoverDeliveries(context.Background(), run.atStart)

			task := requireQueued(t, s.Engine.ExportRetryCh())
			assert.Equal(t, id, task.DeliveryID)
			assert.Equal(t, 2, task.AttemptNum)
			iAssertStatus(t, s.WebhookDB, id, database.DeliveryStatusRetrying)
		})
	}
}

// TestRecovery_RetryingBackoffNotPassed: a retrying delivery whose last
// attempt was just now waits out the rest of its backoff, ten attempts
// in, eight and a half minutes. It is not sent now, and the engine holds
// it, so a later tick leaves it to the timer.
func TestRecovery_RetryingBackoffNotPassed(t *testing.T) {
	t.Parallel()

	for _, run := range recoveryRuns() {
		t.Run(run.name, func(t *testing.T) {
			t.Parallel()

			s, targetID := recoverySetup(t)
			id := seedRecoveryDelivery(
				t, s, targetID, database.DeliveryStatusRetrying, 10, false,
			)

			s.Engine.ExportRecoverDeliveries(context.Background(), run.atStart)

			assertNothingQueued(t, s.Engine)
			iAssertStatus(t, s.WebhookDB, id, database.DeliveryStatusRetrying)
			assert.Equal(t, 1, s.Engine.ExportInflightHeld(),
				"no retry timer holds the delivery",
			)
		})
	}
}

// TestRecovery_PendingOlderThanTheAgeBound: a pending delivery untouched
// for longer than the sweep's age bound is sent by either run.
func TestRecovery_PendingOlderThanTheAgeBound(t *testing.T) {
	t.Parallel()

	for _, run := range recoveryRuns() {
		t.Run(run.name, func(t *testing.T) {
			t.Parallel()

			s, targetID := recoverySetup(t)
			id := seedRecoveryDelivery(
				t, s, targetID, database.DeliveryStatusPending, 0, false,
			)
			rAgePending(t, s.WebhookDB, id)

			s.Engine.ExportRecoverDeliveries(context.Background(), run.atStart)

			task := requireQueued(t, s.Engine.ExportDeliveryCh())
			assert.Equal(t, id, task.DeliveryID)
			assert.Equal(t, 1, task.AttemptNum)
		})
	}
}

// TestRecovery_TaskCarriesEveryTargetSetting: the task the recovery
// routine queues carries every one of the target's settings, all six
// set here.
// They are copied into the task one field at a time, and a field left
// out is sent empty without any error.
func TestRecovery_TaskCarriesEveryTargetSetting(t *testing.T) {
	t.Parallel()

	for _, run := range recoveryRuns() {
		t.Run(run.name, func(t *testing.T) {
			t.Parallel()

			s := newISetup(t)
			iCreateWebhook(t, s.MainDB, s.WebhookID, "recovery")

			target := database.Target{
				WebhookID:       s.WebhookID,
				Name:            "recovery",
				Type:            database.TargetTypeHTTP,
				Active:          true,
				URL:             "http://example.com/hook",
				Headers:         map[string]string{headerAuthorization: "Bearer t"},
				TimeoutSeconds:  7,
				ForwardQuery:    true,
				ArchiveExpiry:   time.Hour,
				ArchiveRotation: "daily",
				MaxAttempts:     1,
			}
			require.NoError(t, s.MainDB.Create(&target).Error)

			id := seedRecoveryDelivery(
				t, s, target.ID, database.DeliveryStatusPending, 0, false,
			)
			rAgePending(t, s.WebhookDB, id)

			s.Engine.ExportRecoverDeliveries(context.Background(), run.atStart)

			task := requireQueued(t, s.Engine.ExportDeliveryCh())
			assert.Equal(t, target.ID, task.TargetID)
			assert.Equal(t, target.URL, task.TargetURL)
			assert.Equal(t, target.Headers, task.TargetHeaders)
			assert.Equal(t, target.TimeoutSeconds, task.TimeoutSeconds)
			assert.Equal(t, target.ForwardQuery, task.ForwardQuery)
			assert.Equal(t, target.ArchiveExpiry, task.ArchiveExpiry)
			assert.Equal(t, target.ArchiveRotation, task.ArchiveRotation)
		})
	}
}

// TestRecovery_PendingYoungerThanTheAgeBound: a pending delivery touched
// a moment ago, as one a stop cut off is, is sent at start, which is how
// it reaches its target on the next start
// (https://git.eeqj.de/sneak/webhooker/issues/527). A tick leaves it.
func TestRecovery_PendingYoungerThanTheAgeBound(t *testing.T) {
	t.Parallel()

	for _, run := range recoveryRuns() {
		t.Run(run.name, func(t *testing.T) {
			t.Parallel()

			s, targetID := recoverySetup(t)
			id := seedRecoveryDelivery(
				t, s, targetID, database.DeliveryStatusPending, 0, false,
			)

			s.Engine.ExportRecoverDeliveries(context.Background(), run.atStart)

			if run.atStart {
				task := requireQueued(t, s.Engine.ExportDeliveryCh())
				assert.Equal(t, id, task.DeliveryID)

				return
			}

			assertNothingQueued(t, s.Engine)
			iAssertStatus(t, s.WebhookDB, id, database.DeliveryStatusPending)
			assert.Zero(t, s.Engine.ExportInflightHeld())
		})
	}
}

// TestRecovery_AlreadySucceeded: a pending or retrying delivery whose
// success is recorded but whose status write was lost is marked
// delivered where it stands, and not sent again.
func TestRecovery_AlreadySucceeded(t *testing.T) {
	t.Parallel()

	for _, run := range recoveryRuns() {
		t.Run(run.name, func(t *testing.T) {
			t.Parallel()

			s, targetID := recoverySetup(t)
			pending := seedRecoveryDelivery(
				t, s, targetID, database.DeliveryStatusPending, 1, true,
			)
			rAgePending(t, s.WebhookDB, pending)

			retrying := seedRecoveryDelivery(
				t, s, targetID, database.DeliveryStatusRetrying, 2, true,
			)
			backdateAttempts(t, s, retrying)

			s.Engine.ExportRecoverDeliveries(context.Background(), run.atStart)

			assertNothingQueued(t, s.Engine)

			for id, attempts := range map[string]int{pending: 1, retrying: 2} {
				iAssertStatus(
					t, s.WebhookDB, id, database.DeliveryStatusDelivered,
				)
				assert.Len(t, iResults(t, s.WebhookDB, id), attempts,
					"marking the delivery delivered recorded an attempt",
				)
			}
		})
	}
}

// TestRecovery_MissingDatabase: a tick leaves a webhook whose database
// file is missing alone, and the run at start opens it, which creates an
// empty one and reports the loss (see
// TestRecoverDeliveries_ReportsAMissingWebhookDatabase).
func TestRecovery_MissingDatabase(t *testing.T) {
	t.Parallel()

	mainDB := iMainDB(t)
	webhookID := uuid.New().String()
	iCreateWebhook(t, mainDB.DB(), webhookID, "missing-database")

	dbMgr := databasetest.NewWebhookDBManager(t, t.TempDir())
	t.Cleanup(func() { _ = dbMgr.CloseAll() })

	engine := delivery.NewTestEngineWithDB(
		mainDB, dbMgr,
		slog.New(slog.DiscardHandler),
		&http.Client{Timeout: 5 * time.Second}, 1,
	)

	engine.ExportRecoverDeliveries(context.Background(), onTick)
	assert.False(t, dbMgr.DBExists(webhookID),
		"a tick created the missing database",
	)

	engine.ExportRecoverDeliveries(context.Background(), atStart)
	assert.True(t, dbMgr.DBExists(webhookID),
		"the run at start did not open the missing database",
	)
}

// TestRecovery_LeavesADeliveryTheEngineHolds: a delivery the engine is
// working on, has queued, or holds a retry timer for is left alone by
// either run, pending and retrying alike, however old.
func TestRecovery_LeavesADeliveryTheEngineHolds(t *testing.T) {
	t.Parallel()

	for _, run := range recoveryRuns() {
		t.Run(run.name, func(t *testing.T) {
			t.Parallel()

			s, targetID := recoverySetup(t)
			pending := seedRecoveryDelivery(
				t, s, targetID, database.DeliveryStatusPending, 0, false,
			)
			rAgePending(t, s.WebhookDB, pending)

			retrying := seedRecoveryDelivery(
				t, s, targetID, database.DeliveryStatusRetrying, 1, false,
			)
			backdateAttempts(t, s, retrying)

			require.True(t, s.Engine.ExportRetainDelivery(pending))
			require.True(t, s.Engine.ExportRetainDelivery(retrying))

			s.Engine.ExportRecoverDeliveries(context.Background(), run.atStart)

			assertNothingQueued(t, s.Engine)
			iAssertStatus(t, s.WebhookDB, pending, database.DeliveryStatusPending)
			iAssertStatus(t, s.WebhookDB, retrying, database.DeliveryStatusRetrying)
			assert.Equal(t, 2, s.Engine.ExportInflightHeld())
		})
	}
}