276 lines
8.4 KiB
Go
276 lines
8.4 KiB
Go
package httpserver_test
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/krow/krow-backend/go-api/internal/httpserver"
|
|
)
|
|
|
|
// Scheduler lifecycle.
|
|
//
|
|
// What is under test is the GOROUTINE, not the deletes — those are covered in
|
|
// internal/oauth and internal/ratelimit against real data. Here the questions
|
|
// are: does it start, does it do a pass, does it stop when told, does a failure
|
|
// take the process with it, and is running it twice safe.
|
|
|
|
/* ── Lifecycle ──────────────────────────────────────────────────────────── */
|
|
|
|
// It runs one pass IMMEDIATELY, before the first tick. A process that has been
|
|
// down should not carry a backlog for a further hour.
|
|
func TestMaintenanceRunsOnceImmediately(t *testing.T) {
|
|
a := newOAuthAPI(t)
|
|
m := a.srv.Maintenance()
|
|
if m == nil {
|
|
t.Fatal("a configured deployment returned no Maintenance")
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
httpserver.SweepMaintenance(ctx, m, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
close(done)
|
|
}()
|
|
|
|
// The immediate pass is the only one that will happen inside the test's
|
|
// lifetime — the ticker is an hour. Give it a moment, then stop.
|
|
time.Sleep(200 * time.Millisecond)
|
|
cancel()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the sweeper did not stop within 5s of cancellation")
|
|
}
|
|
}
|
|
|
|
// Cancellation must return promptly, or a shutdown hangs on a goroutine nobody
|
|
// is waiting for.
|
|
func TestMaintenanceStopsOnCancellation(t *testing.T) {
|
|
a := newOAuthAPI(t)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
httpserver.SweepMaintenance(ctx, a.srv.Maintenance(),
|
|
slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
close(done)
|
|
}()
|
|
|
|
time.Sleep(100 * time.Millisecond)
|
|
start := time.Now()
|
|
cancel()
|
|
|
|
select {
|
|
case <-done:
|
|
if elapsed := time.Since(start); elapsed > 2*time.Second {
|
|
t.Errorf("stopping took %v; shutdown would block on it", elapsed)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the sweeper ignored cancellation")
|
|
}
|
|
}
|
|
|
|
// An already-cancelled context must not run a pass and must return at once.
|
|
func TestMaintenanceWithAnAlreadyCancelledContextReturns(t *testing.T) {
|
|
a := newOAuthAPI(t)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
httpserver.SweepMaintenance(ctx, a.srv.Maintenance(),
|
|
slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the sweeper did not return on an already-cancelled context")
|
|
}
|
|
}
|
|
|
|
/* ── It does the work ───────────────────────────────────────────────────── */
|
|
|
|
// One pass removes dead rows and leaves live ones. The detailed retention rules
|
|
// are tested in internal/oauth; this asserts the scheduler is wired to them.
|
|
func TestMaintenanceSweepRemovesDeadRows(t *testing.T) {
|
|
a := newOAuthAPI(t)
|
|
ctx := context.Background()
|
|
|
|
// A grant that is already past its expiry and its grace.
|
|
if _, err := a.h.Pool.Exec(ctx,
|
|
`INSERT INTO oauth_clients (client_id, client_name, redirect_uris)
|
|
VALUES ('sweep-client', 'Sweep', ARRAY['https://a.test/cb'])`); err != nil {
|
|
t.Fatalf("client: %v", err)
|
|
}
|
|
userID, _ := seededUser(t, a.h.Pool)
|
|
var orgID string
|
|
if err := a.h.Pool.QueryRow(ctx,
|
|
`SELECT org_id::text FROM users WHERE id = $1::uuid`, userID).Scan(&orgID); err != nil {
|
|
t.Fatalf("org: %v", err)
|
|
}
|
|
|
|
if _, err := a.h.Pool.Exec(ctx,
|
|
`INSERT INTO oauth_grants
|
|
(code_hash, client_id, user_id, org_id, redirect_uri, scopes, resource,
|
|
code_challenge, code_challenge_method, created_date, expires_at)
|
|
VALUES (repeat('a', 64), 'sweep-client', $1::uuid, $2::uuid, 'https://a.test/cb',
|
|
ARRAY['krow.read'], $3, repeat('B', 43), 'S256',
|
|
now() - interval '3 hours', now() - interval '3 hours' + interval '1 minute')`,
|
|
userID, orgID, testMCPResource); err != nil {
|
|
t.Fatalf("grant: %v", err)
|
|
}
|
|
|
|
// An expired rate-limit bucket.
|
|
if _, err := a.h.Pool.Exec(ctx,
|
|
`INSERT INTO rate_limits (bucket, window_start, count, expires_at)
|
|
VALUES ('test:old', now() - interval '3 hours', 5, now() - interval '2 hours')`); err != nil {
|
|
t.Fatalf("bucket: %v", err)
|
|
}
|
|
// And a live one, which must survive.
|
|
if _, err := a.h.Pool.Exec(ctx,
|
|
`INSERT INTO rate_limits (bucket, window_start, count, expires_at)
|
|
VALUES ('test:live', now(), 1, now() + interval '1 hour')`); err != nil {
|
|
t.Fatalf("bucket: %v", err)
|
|
}
|
|
|
|
result, err := a.srv.Maintenance().Sweep(ctx)
|
|
if err != nil {
|
|
t.Fatalf("Sweep: %v", err)
|
|
}
|
|
|
|
if result.Grants != 1 {
|
|
t.Errorf("removed %d grants, want 1", result.Grants)
|
|
}
|
|
if result.RateLimits != 1 {
|
|
t.Errorf("removed %d rate-limit rows, want 1", result.RateLimits)
|
|
}
|
|
if result.Total() != 2 {
|
|
t.Errorf("Total() = %d, want 2", result.Total())
|
|
}
|
|
|
|
var live int
|
|
if err := a.h.Pool.QueryRow(ctx,
|
|
`SELECT count(*) FROM rate_limits WHERE bucket = 'test:live'`).Scan(&live); err != nil {
|
|
t.Fatalf("count: %v", err)
|
|
}
|
|
if live != 1 {
|
|
t.Error("the live rate-limit window was swept")
|
|
}
|
|
}
|
|
|
|
// Running it repeatedly must be safe and must converge to removing nothing.
|
|
func TestRepeatedMaintenanceIsSafe(t *testing.T) {
|
|
a := newOAuthAPI(t)
|
|
ctx := context.Background()
|
|
m := a.srv.Maintenance()
|
|
|
|
for i := 0; i < 3; i++ {
|
|
result, err := m.Sweep(ctx)
|
|
if err != nil {
|
|
t.Fatalf("pass %d: %v", i+1, err)
|
|
}
|
|
if i > 0 && result.Total() != 0 {
|
|
t.Errorf("pass %d removed %d rows; a repeat pass should find nothing", i+1, result.Total())
|
|
}
|
|
}
|
|
}
|
|
|
|
// Two instances sweep concurrently with no coordination. Neither may error.
|
|
// Run with -race.
|
|
func TestConcurrentMaintenanceIsSafe(t *testing.T) {
|
|
a := newOAuthAPI(t)
|
|
ctx := context.Background()
|
|
|
|
const instances = 4
|
|
var wg sync.WaitGroup
|
|
errs := make(chan error, instances)
|
|
|
|
for i := 0; i < instances; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if _, err := a.srv.Maintenance().Sweep(ctx); err != nil {
|
|
errs <- err
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
close(errs)
|
|
|
|
for err := range errs {
|
|
t.Errorf("concurrent sweep errored: %v", err)
|
|
}
|
|
}
|
|
|
|
/* ── Failure isolation ──────────────────────────────────────────────────── */
|
|
|
|
// A failing sweep must be logged and survived, not fatal. The database is
|
|
// closed underneath the sweeper, which is the closest thing to a real outage a
|
|
// test can arrange.
|
|
func TestMaintenanceSurvivesADatabaseFailure(t *testing.T) {
|
|
a := newOAuthAPI(t)
|
|
m := a.srv.Maintenance()
|
|
|
|
var logged strings.Builder
|
|
log := slog.New(slog.NewTextHandler(&logged, &slog.HandlerOptions{Level: slog.LevelDebug}))
|
|
|
|
// A cancelled context makes every statement fail immediately.
|
|
dead, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
|
|
if _, err := m.Sweep(dead); err == nil {
|
|
t.Log("note: the sweep reported no error on a cancelled context")
|
|
}
|
|
|
|
// The goroutine wrapper must not panic or exit the process on that.
|
|
ctx, stop := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() {
|
|
httpserver.SweepMaintenance(ctx, m, log)
|
|
close(done)
|
|
}()
|
|
time.Sleep(150 * time.Millisecond)
|
|
stop()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the sweeper did not stop")
|
|
}
|
|
}
|
|
|
|
/* ── It is absent when the surface is ───────────────────────────────────── */
|
|
|
|
// A deployment without OAuth has nothing to sweep, and must not start a ticker
|
|
// that runs for the life of the process doing nothing.
|
|
func TestMaintenanceIsNilWhenTheSurfaceIsDisabled(t *testing.T) {
|
|
a := newAPI(t) // the standard fixture: no OAuth configuration
|
|
|
|
if m := a.srv.Maintenance(); m != nil {
|
|
t.Error("an unconfigured deployment returned a Maintenance sweeper")
|
|
}
|
|
|
|
// And the runner must return immediately rather than tick forever.
|
|
done := make(chan struct{})
|
|
go func() {
|
|
httpserver.SweepMaintenance(context.Background(), a.srv.Maintenance(),
|
|
slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("the sweeper ticked despite having nothing to sweep")
|
|
}
|
|
}
|