dnswatcher/internal/notify/shutdown_test.go

package notify_test

import (
	"bytes"
	"context"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"go.uber.org/fx"

	"sneak.berlin/go/dnswatcher/internal/config"
	"sneak.berlin/go/dnswatcher/internal/globals"
	"sneak.berlin/go/dnswatcher/internal/logger"
	"sneak.berlin/go/dnswatcher/internal/notify"
)

// Timings used by the drain tests.  They stay in the same
// 10-100ms band as the retry tests so the suite never waits on
// a real backoff delay.
const (
	// inFlightHold is how long a delivery is kept mid-request
	// before the handler is released.
	inFlightHold = 30 * time.Millisecond

	// drainDeadline bounds a drain that is expected to time
	// out.
	drainDeadline = 50 * time.Millisecond

	// drainSlack is the upper bound on how long a bounded
	// drain may take; generous enough for a loaded CI box,
	// still far below the 20s test ceiling.
	drainSlack = 2 * time.Second

	// settleDelay is how long to wait before asserting that
	// something did *not* happen.
	settleDelay = 50 * time.Millisecond
)

// syncBuffer is an io.Writer safe for concurrent use, so log
// output written from delivery goroutines can be inspected.
type syncBuffer struct {
	mu  sync.Mutex
	buf bytes.Buffer
}

func (sb *syncBuffer) Write(p []byte) (int, error) {
	sb.mu.Lock()
	defer sb.mu.Unlock()

	return sb.buf.Write(p) //nolint:wrapcheck // test helper
}

func (sb *syncBuffer) String() string {
	sb.mu.Lock()
	defer sb.mu.Unlock()

	return sb.buf.String()
}

// newLoggingService returns a Service writing JSON logs into
// the returned buffer.
func newLoggingService(
	transport http.RoundTripper,
) (*notify.Service, *syncBuffer) {
	logs := &syncBuffer{}
	handler := slog.NewJSONHandler(logs, nil)

	return notify.NewTestServiceWithLogger(transport, handler),
		logs
}

// blockingNtfyServer returns a server whose handler signals on
// entered, waits for release, and then responds 200.
func blockingNtfyServer(
	entered chan<- struct{},
	release <-chan struct{},
	served *atomic.Bool,
) *httptest.Server {
	var once sync.Once

	return httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				once.Do(func() { close(entered) })
				<-release

				served.Store(true)

				w.WriteHeader(http.StatusOK)
			}),
	)
}

// TestDrainWaitsForInFlightDelivery verifies that a delivery
// already under way when shutdown starts is allowed to finish.
func TestDrainWaitsForInFlightDelivery(t *testing.T) {
	t.Parallel()

	var served atomic.Bool

	entered := make(chan struct{})
	release := make(chan struct{})

	srv := blockingNtfyServer(entered, release, &served)
	defer srv.Close()

	topicURL, _ := url.Parse(srv.URL)

	svc := notify.NewTestService(http.DefaultTransport)
	svc.SetNtfyURL(topicURL)

	svc.SendNotification(
		context.Background(), "t", "m", prioInfo,
	)

	// Make sure the delivery really is mid-request before the
	// drain begins.
	select {
	case <-entered:
	case <-time.After(drainSlack):
		t.Fatal("delivery never reached the endpoint")
	}

	timer := time.AfterFunc(inFlightHold, func() {
		close(release)
	})
	defer timer.Stop()

	ctx, cancel := context.WithTimeout(
		context.Background(), drainSlack,
	)
	defer cancel()

	start := time.Now()

	svc.Drain(ctx)

	elapsed := time.Since(start)

	if !served.Load() {
		t.Error(
			"drain returned before the in-flight delivery " +
				"completed",
		)
	}

	if elapsed < inFlightHold {
		t.Errorf(
			"drain took %v, want at least %v",
			elapsed, inFlightHold,
		)
	}

	if got := svc.OutstandingDeliveries(); got != 0 {
		t.Errorf("outstanding deliveries = %d, want 0", got)
	}
}

// neverFires returns a channel that never delivers, standing in
// for a long backoff sleep without actually sleeping.
func neverFires(_ time.Duration) <-chan time.Time {
	return make(chan time.Time)
}

// TestDrainBoundedByContextDeadline verifies that a delivery
// stuck retrying against a dead endpoint does not hold shutdown
// past the OnStop context deadline, and that the abandoned
// deliveries are logged at warn level rather than dropped
// silently.
func TestDrainBoundedByContextDeadline(t *testing.T) {
	t.Parallel()

	var requests atomic.Int64

	srv := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				requests.Add(1)

				w.WriteHeader(http.StatusInternalServerError)
			}),
	)
	defer srv.Close()

	topicURL, _ := url.Parse(srv.URL)

	svc, logs := newLoggingService(http.DefaultTransport)
	svc.SetNtfyURL(topicURL)
	// Never let the backoff sleep complete: the delivery is
	// parked in its retry wait until shutdown releases it.
	svc.SetSleepFunc(neverFires)
	svc.SetRetryConfig(notify.RetryConfig{
		MaxRetries: 5,
		BaseDelay:  time.Hour,
		MaxDelay:   time.Hour,
	})

	svc.SendNotification(
		context.Background(), "t", "m", prioError,
	)

	waitForCondition(t, func() bool {
		return requests.Load() >= 1 &&
			svc.OutstandingDeliveries() == 1
	})

	ctx, cancel := context.WithTimeout(
		context.Background(), drainDeadline,
	)
	defer cancel()

	start := time.Now()

	svc.Drain(ctx)

	elapsed := time.Since(start)

	if elapsed < drainDeadline {
		t.Errorf(
			"drain returned after %v, before its %v deadline",
			elapsed, drainDeadline,
		)
	}

	if elapsed > drainSlack {
		t.Errorf(
			"drain took %v, want it bounded well under %v",
			elapsed, drainSlack,
		)
	}

	assertAbandonLogged(t, logs.String())

	// The abandoned delivery must stop retrying rather than
	// outlive the drain.
	waitForCondition(t, func() bool {
		return svc.OutstandingDeliveries() == 0
	})
}

// assertAbandonLogged checks that the drain logged the
// abandoned deliveries at warn level with a count.
func assertAbandonLogged(t *testing.T, output string) {
	t.Helper()

	if !strings.Contains(output, `"level":"WARN"`) {
		t.Errorf(
			"abandoned deliveries not logged at warn level; "+
				"log output: %s",
			output,
		)
	}

	if !strings.Contains(output, `"abandoned":1`) {
		t.Errorf(
			"abandoned delivery count not logged; "+
				"log output: %s",
			output,
		)
	}
}

// TestDrainRefusesNewDeliveries verifies that notifications
// submitted after the drain has begun are refused and logged,
// so a stream of new work cannot extend shutdown indefinitely.
func TestDrainRefusesNewDeliveries(t *testing.T) {
	t.Parallel()

	var requests atomic.Int64

	srv := httptest.NewServer(
		http.HandlerFunc(
			func(w http.ResponseWriter, _ *http.Request) {
				requests.Add(1)

				w.WriteHeader(http.StatusOK)
			}),
	)
	defer srv.Close()

	target, _ := url.Parse(srv.URL)

	svc, logs := newLoggingService(http.DefaultTransport)
	svc.SetNtfyURL(target)
	svc.SetSlackWebhookURL(target)
	svc.SetMattermostWebhookURL(target)

	ctx, cancel := context.WithTimeout(
		context.Background(), drainSlack,
	)
	defer cancel()

	// Nothing is in flight, so this returns immediately and
	// leaves the service refusing further deliveries.
	svc.Drain(ctx)

	for range 3 {
		svc.SendNotification(
			context.Background(), "t", "m", prioInfo,
		)
	}

	time.Sleep(settleDelay)

	if got := requests.Load(); got != 0 {
		t.Errorf(
			"%d requests reached the endpoint after drain, "+
				"want 0",
			got,
		)
	}

	if got := svc.OutstandingDeliveries(); got != 0 {
		t.Errorf("outstanding deliveries = %d, want 0", got)
	}

	output := logs.String()
	if !strings.Contains(output, "shutdown in progress") {
		t.Errorf(
			"refused deliveries not logged; log output: %s",
			output,
		)
	}
}

// recordingLifecycle is a minimal fx.Lifecycle that records the
// hooks appended to it, so the wiring done by notify.New can be
// inspected without standing up a whole fx application.
type recordingLifecycle struct {
	hooks []fx.Hook
}

func (l *recordingLifecycle) Append(hook fx.Hook) {
	l.hooks = append(l.hooks, hook)
}

// newNotifyService builds a Service through the real
// constructor, wired to the given lifecycle.
func newNotifyService(
	t *testing.T,
	lifecycle fx.Lifecycle,
	ntfyTopic string,
) *notify.Service {
	t.Helper()

	g, err := globals.New(nil)
	if err != nil {
		t.Fatalf("globals.New: %v", err)
	}

	log, err := logger.New(nil, logger.Params{Globals: g})
	if err != nil {
		t.Fatalf("logger.New: %v", err)
	}

	svc, err := notify.New(lifecycle, notify.Params{
		Logger: log,
		Config: &config.Config{NtfyTopic: ntfyTopic},
	})
	if err != nil {
		t.Fatalf("notify.New: %v", err)
	}

	return svc
}

// TestNewRegistersDrainingStopHook verifies that notify.New
// wires an OnStop hook into the fx lifecycle and that the hook
// waits for in-flight deliveries.
func TestNewRegistersDrainingStopHook(t *testing.T) {
	t.Parallel()

	var served atomic.Bool

	entered := make(chan struct{})
	release := make(chan struct{})

	srv := blockingNtfyServer(entered, release, &served)
	defer srv.Close()

	lifecycle := &recordingLifecycle{}
	svc := newNotifyService(t, lifecycle, srv.URL)

	if len(lifecycle.hooks) != 1 {
		t.Fatalf(
			"appended %d lifecycle hooks, want 1",
			len(lifecycle.hooks),
		)
	}

	stop := lifecycle.hooks[0].OnStop
	if stop == nil {
		t.Fatal("lifecycle hook has no OnStop function")
	}

	svc.SendNotification(
		context.Background(), "t", "m", prioInfo,
	)

	select {
	case <-entered:
	case <-time.After(drainSlack):
		t.Fatal("delivery never reached the endpoint")
	}

	timer := time.AfterFunc(inFlightHold, func() {
		close(release)
	})
	defer timer.Stop()

	ctx, cancel := context.WithTimeout(
		context.Background(), drainSlack,
	)
	defer cancel()

	err := stop(ctx)
	if err != nil {
		t.Fatalf("OnStop returned error: %v", err)
	}

	if !served.Load() {
		t.Error(
			"OnStop returned before the in-flight delivery " +
				"completed",
		)
	}
}

// TestDrainWithoutDeliveriesReturnsImmediately verifies the
// common case: nothing in flight, shutdown is not delayed.
func TestDrainWithoutDeliveriesReturnsImmediately(t *testing.T) {
	t.Parallel()

	svc := notify.NewTestService(http.DefaultTransport)

	ctx, cancel := context.WithTimeout(
		context.Background(), drainSlack,
	)
	defer cancel()

	start := time.Now()

	svc.Drain(ctx)

	if elapsed := time.Since(start); elapsed > settleDelay {
		t.Errorf(
			"drain of an idle service took %v, want ~0",
			elapsed,
		)
	}
}