test: expand risk-based coverage

internal/app/resumable_coverage_extra_test.go

@@ -26,6 +26,32 @@ type coverageResumableScanner struct {
 	clearLeases bool
 }
 
+type lateSuccessScanner struct {
+	cancel context.CancelFunc
+}
+
+func (s lateSuccessScanner) Version(context.Context) string { return "late-success" }
+func (s lateSuccessScanner) Scan(context.Context, config.Job) (model.Snapshot, error) {
+	return model.Snapshot{}, errors.New("ordinary scan path is not expected")
+}
+func (s lateSuccessScanner) Plan(context.Context, config.Job) (scanner.WorkPlan, error) {
+	return scanner.WorkPlan{Units: []scanner.WorkUnit{
+		{Sequence: 0, Protocol: "tcp", Addresses: []string{"192.0.2.1"}, Ports: "1", PortCount: 1, Probes: 1},
+		{Sequence: 1, Protocol: "tcp", Addresses: []string{"192.0.2.1"}, Ports: "2", PortCount: 1, Probes: 1},
+	}}, nil
+}
+func (s lateSuccessScanner) ScanWorkUnit(_ context.Context, _ config.Job, unit scanner.WorkUnit, _ scanner.ProgressReporter) (model.Snapshot, error) {
+	if s.cancel != nil {
+		s.cancel()
+		return model.Snapshot{Units: []model.Unit{{Target: "192.0.2.1", Protocol: "tcp", Addresses: unit.Addresses, Ports: []model.PortState{{Port: 999, State: "open"}}}}}, nil
+	}
+	port := 443
+	if unit.Sequence == 1 {
+		port = 80
+	}
+	return model.Snapshot{Units: []model.Unit{{Target: "192.0.2.1", Protocol: "tcp", Addresses: unit.Addresses, Ports: []model.PortState{{Port: port, State: "open"}}}}}, nil
+}
+
 type transientResumableScanner struct {
 	mu         sync.Mutex
 	calls      map[int]int
@@ -134,6 +160,56 @@ func TestResumableScanRetriesTransientUnitFailure(t *testing.T) {
 	}
 }
 
+func TestResumableAttemptDiscardsLateSuccessfulResultAfterCancellation(t *testing.T) {
+	ctx := context.Background()
+	db, err := store.Open(storetest.FreshPath(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	cfg := &config.Config{Version: 1, Database: db.Path, Retention: config.Duration(24 * time.Hour), Scheduler: config.Scheduler{MaxConcurrent: 1}, Web: config.Web{Listen: "127.0.0.1:8080"}}
+	a, err := New(cfg, db, "missing-nmap", slog.New(slog.NewTextHandler(io.Discard, nil)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	job := config.NormalizeJob(config.Job{Name: "late-success", Schedule: "0 * * * *", Timezone: "UTC", Targets: []string{"192.0.2.1"}, TCP: &config.Protocol{Ports: "1-2", Mode: "connect"}, Timeout: config.Duration(time.Minute), ResumeWindow: config.Duration(time.Hour)})
+	record, err := defaultTenant(db).CreateJob(ctx, job)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	attemptCtx, cancel := context.WithCancel(ctx)
+	var interrupted model.Scan
+	handled, lateFragment, runErr := a.runResumableAttempt(ctx, attemptCtx, defaultTenant(db), job, record.ID, &interrupted, nil, lateSuccessScanner{cancel: cancel}, false)
+	if !handled || !errors.Is(runErr, context.Canceled) || interrupted.Status != "canceled" || !interrupted.Resumable || interrupted.CycleStatus != "paused" {
+		t.Fatalf("late successful return = handled %v scan %#v err %v", handled, interrupted, runErr)
+	}
+	if len(lateFragment.Units) != 1 || lateFragment.Units[0].Ports[0].Port != 999 {
+		t.Fatalf("scanner fixture did not return its late fragment: %#v", lateFragment)
+	}
+	cycle, err := defaultTenant(db).GetActiveScanCycle(ctx, record.ID)
+	if err != nil || cycle.Status != "paused" || cycle.CompletedUnits != 0 {
+		t.Fatalf("cycle after canceled late return = %#v, %v", cycle, err)
+	}
+	summaries, err := defaultTenant(db).ListScanCycleUnitSummaries(ctx, cycle.ID)
+	if err != nil || len(summaries) != 2 || summaries[0].Status != "pending" || summaries[1].Status != "pending" {
+		t.Fatalf("late result was checkpointed: %#v, %v", summaries, err)
+	}
+	_, checkpoints, err := db.System().LoadScanCycleFragments(ctx, cycle.ID)
+	if err != nil || len(checkpoints) != 0 {
+		t.Fatalf("canceled attempt fragments = %#v, %v", checkpoints, err)
+	}
+
+	var resumed model.Scan
+	handled, snapshot, runErr := a.runResumableAttempt(ctx, ctx, defaultTenant(db), job, record.ID, &resumed, nil, lateSuccessScanner{}, false)
+	if !handled || runErr != nil || resumed.Status != "success" || resumed.CycleStatus != "completed" {
+		t.Fatalf("resumed attempt = handled %v scan %#v err %v", handled, resumed, runErr)
+	}
+	if len(snapshot.Units) != 1 || len(snapshot.Units[0].Ports) != 2 || snapshot.Units[0].Ports[0].Port != 80 || snapshot.Units[0].Ports[1].Port != 443 {
+		t.Fatalf("resumed snapshot included the discarded late result: %#v", snapshot)
+	}
+}
+
 func TestResumableScanStallsAfterRetryBudget(t *testing.T) {
 	ctx := context.Background()
 	db, err := store.Open(storetest.FreshPath(t))

internal/store/latest_projection_coverage_test.go

@@ -40,3 +40,54 @@ func TestRebuildLatestScanHostsWrapperRecreatesProjection(t *testing.T) {
 		t.Fatalf("canceled rebuild error = %v, want context.Canceled", err)
 	}
 }
+
+func TestRepairLatestScanHostsCancellationAndRollback(t *testing.T) {
+	ctx := context.Background()
+	s := openTestStore(t)
+
+	canceled, cancel := context.WithCancel(ctx)
+	cancel()
+	if err := s.repairLatestScanHosts(canceled); !errors.Is(err, context.Canceled) {
+		t.Fatalf("canceled repair error = %v, want context.Canceled", err)
+	}
+
+	insertJobRows(t, s, "repair-rollback")
+	address := fixtureHost(0).Address
+	base := time.Date(2026, time.September, 20, 12, 0, 0, 0, time.UTC)
+	for _, scan := range []model.Scan{
+		fixtureScan("repair-old", "repair-rollback", "repair-rollback", base, []model.HostObservation{fixtureHost(0)}),
+		fixtureScan("repair-new", "repair-rollback", "repair-rollback", base.Add(time.Minute), []model.HostObservation{fixtureHost(0)}),
+	} {
+		if err := s.System().SaveScan(ctx, scan); err != nil {
+			t.Fatalf("save %s: %v", scan.ID, err)
+		}
+	}
+	if _, err := s.DB.ExecContext(ctx, `DELETE FROM scans WHERE id='repair-new'`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.DB.ExecContext(ctx, `CREATE TRIGGER reject_latest_host_repair BEFORE INSERT ON latest_scan_hosts BEGIN SELECT RAISE(ABORT, 'temporary projection failure'); END`); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.repairLatestScanHosts(ctx); err == nil {
+		t.Fatal("repair insert failure was ignored")
+	}
+	var scanID string
+	if err := s.DB.QueryRowContext(ctx, `SELECT scan_id FROM latest_scan_hosts WHERE tenant_id=? AND address=?`, DefaultTenantID, address).Scan(&scanID); err != nil {
+		t.Fatal(err)
+	}
+	if scanID != "repair-new" {
+		t.Fatalf("failed repair partially deleted the projection row: scan_id=%q", scanID)
+	}
+	if _, err := s.DB.ExecContext(ctx, `DROP TRIGGER reject_latest_host_repair`); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.repairLatestScanHosts(ctx); err != nil {
+		t.Fatalf("repair after removing transient failure: %v", err)
+	}
+	if err := s.DB.QueryRowContext(ctx, `SELECT scan_id FROM latest_scan_hosts WHERE tenant_id=? AND address=?`, DefaultTenantID, address).Scan(&scanID); err != nil {
+		t.Fatal(err)
+	}
+	if scanID != "repair-old" {
+		t.Fatalf("repaired projection scan = %q, want retained older scan", scanID)
+	}
+}

internal/store/sse_cursor_test.go

@@ -49,3 +49,30 @@ func TestReserveSSEEventIDsSurvivesRestartAndFollowsDurableEvents(t *testing.T)
 		t.Fatalf("durable SSE cursor = %d, want %d", cursor, nextEnd)
 	}
 }
+
+func TestStoreSSECursorReservesGlobalRanges(t *testing.T) {
+	ctx := context.Background()
+	db := openTestStore(t)
+	cursor := db.SSECursor()
+
+	start, end, err := cursor.Reserve(ctx, 2)
+	if err != nil || start != 1 || end != 2 {
+		t.Fatalf("first cursor range = %d-%d, %v; want 1-2", start, end, err)
+	}
+	start, end, err = cursor.ReserveAfter(ctx, 3, 10)
+	if err != nil || start != 11 || end != 13 {
+		t.Fatalf("range after future marker = %d-%d, %v; want 11-13", start, end, err)
+	}
+
+	maxEventID, err := cursor.MaxEventID(ctx)
+	if err != nil || maxEventID != 0 {
+		t.Fatalf("event high-water mark = %d, %v; want 0 because reservations are not events", maxEventID, err)
+	}
+	var nextID int64
+	if err := db.DB.QueryRowContext(ctx, `SELECT next_id FROM sse_event_cursor WHERE id=1`).Scan(&nextID); err != nil {
+		t.Fatal(err)
+	}
+	if nextID != 13 {
+		t.Fatalf("durable next ID = %d, want 13", nextID)
+	}
+}

internal/web/server_coverage_extra_test.go

@@ -3,8 +3,10 @@ package web
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -460,3 +462,118 @@ func TestSSECursorReservationRecoversAfterStartupFailure(t *testing.T) {
 		t.Fatalf("durable cursor=%d is behind recovered event=%d", durableCursor, recoveredID)
 	}
 }
+
+func TestSSEReservationFailureBackoffCapsAndRecoversMonotonically(t *testing.T) {
+	db, err := store.Open(storetest.FreshPath(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.DB.ExecContext(context.Background(), `CREATE TRIGGER fail_sse_reservation BEFORE UPDATE ON sse_event_cursor BEGIN SELECT RAISE(ABORT, 'temporary cursor failure'); END`); err != nil {
+		t.Fatal(err)
+	}
+	server := NewServer(nil, db, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	now := time.Date(2026, time.October, 1, 12, 0, 0, 0, time.UTC)
+	server.now = func() time.Time { return now }
+	server.mu.Lock()
+	server.sseDurable = false
+	server.sseRetryAt = now.Add(-time.Second)
+	server.nextEventID = 100
+	server.eventIDLimit = 99
+	server.mu.Unlock()
+
+	server.retrySSEReservationContext(context.Background())
+	server.mu.Lock()
+	if server.sseDurable || server.sseRetryDelay != 2*defaultSSEReservationRetry || !server.sseRetryAt.Equal(now.Add(2*defaultSSEReservationRetry)) || server.nextEventID <= 100 || server.eventIDLimit != ^uint64(0) {
+		got := fmt.Sprintf("durable=%t delay=%s retry=%s next=%d limit=%d", server.sseDurable, server.sseRetryDelay, server.sseRetryAt, server.nextEventID, server.eventIDLimit)
+		server.mu.Unlock()
+		t.Fatalf("first failed reservation state: %s", got)
+	}
+	fallbackID := server.nextEventID
+	server.sseRetryDelay = maxSSEReservationRetry/2 + time.Nanosecond
+	server.mu.Unlock()
+
+	now = now.Add(2 * defaultSSEReservationRetry)
+	server.retrySSEReservationContext(context.Background())
+	server.mu.Lock()
+	if server.sseDurable || server.sseRetryDelay != maxSSEReservationRetry || !server.sseRetryAt.Equal(now.Add(maxSSEReservationRetry)) {
+		got := fmt.Sprintf("durable=%t delay=%s retry=%s", server.sseDurable, server.sseRetryDelay, server.sseRetryAt)
+		server.mu.Unlock()
+		t.Fatalf("capped reservation retry state: %s", got)
+	}
+	server.mu.Unlock()
+
+	if _, err := db.DB.ExecContext(context.Background(), `DROP TRIGGER fail_sse_reservation`); err != nil {
+		t.Fatal(err)
+	}
+	now = now.Add(maxSSEReservationRetry + time.Second)
+	server.retrySSEReservationContext(context.Background())
+	server.mu.Lock()
+	recoveredID, durable, delay, retryAt := server.nextEventID, server.sseDurable, server.sseRetryDelay, server.sseRetryAt
+	server.mu.Unlock()
+	if !durable || recoveredID < fallbackID || delay != 0 || !retryAt.IsZero() {
+		t.Fatalf("recovered cursor state: durable=%t previous=%d recovered=%d delay=%s retry=%s", durable, fallbackID, recoveredID, delay, retryAt)
+	}
+	server.broadcast(map[string]any{"type": "after-reservation-recovery"})
+	server.mu.Lock()
+	monotonicID := server.nextEventID
+	server.mu.Unlock()
+	if monotonicID <= fallbackID {
+		t.Fatalf("post-recovery event ID %d did not advance beyond fallback ID %d", monotonicID, fallbackID)
+	}
+}
+
+func TestListenAndServeEnforcesLoopbackAndJoinsOnShutdown(t *testing.T) {
+	server, _, _ := newUsersTestServer(t)
+	for _, address := range []string{"0.0.0.0:8080", "example.com:8080", "127.0.0.1:0", "not-a-listener"} {
+		if err := server.ListenAndServe(context.Background(), address); err == nil {
+			t.Errorf("ListenAndServe accepted unsafe or invalid address %q", address)
+		}
+	}
+
+	reserved, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := server.ListenAndServe(context.Background(), reserved.Addr().String()); err == nil {
+		t.Fatal("ListenAndServe unexpectedly bound an address already in use")
+	}
+	if err := reserved.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	probe, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	address := probe.Addr().String()
+	if err := probe.Close(); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- server.ListenAndServe(ctx, address) }()
+
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		connection, dialErr := net.DialTimeout("tcp", address, 25*time.Millisecond)
+		if dialErr == nil {
+			_ = connection.Close()
+			break
+		}
+		if time.Now().After(deadline) {
+			cancel()
+			t.Fatalf("loopback listener did not start: %v", dialErr)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("graceful listener shutdown: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("ListenAndServe did not join after context cancellation")
+	}
+}

src/pages/Dashboard.test.tsx

@@ -238,6 +238,43 @@ describe('dashboard', () => {
     expect(container.querySelector('button[aria-label="Run demo"]')).toBeNull()
   })
 
+  it('allows operators to run jobs without exposing administrator notification controls', async () => {
+    vi.mocked(getSession).mockResolvedValue({ ...session('operator'), permissions: ['jobs.read', 'jobs.write', 'scans.read'] })
+    await renderDashboard()
+    expect(container.querySelector('button[aria-label="Run demo"]')).toBeTruthy()
+    expect(container.textContent).toContain('Configure job')
+    expect(container.textContent).not.toContain('Test notifications')
+    expect(notificationTest).not.toHaveBeenCalled()
+  })
+
+  it('restores run, cancel, and notification actions after API failures', async () => {
+    vi.mocked(runJob).mockRejectedValueOnce(new Error('worker unavailable'))
+    vi.mocked(cancelScan).mockRejectedValueOnce(new Error('cancel unavailable'))
+    vi.mocked(notificationTest).mockRejectedValueOnce(new Error('delivery unavailable'))
+    await renderDashboard()
+
+    const run = container.querySelector('button[aria-label="Run demo"]') as HTMLButtonElement
+    await act(async () => { run.click(); await Promise.resolve(); await Promise.resolve() })
+    expect(container.querySelector('[role="alert"]')?.textContent).toContain('worker unavailable')
+    expect(run).toBeEnabled()
+    await act(async () => { run.click(); await Promise.resolve(); await Promise.resolve() })
+    expect(runJob).toHaveBeenCalledTimes(2)
+
+    const cancel = Array.from(container.querySelectorAll('button')).find(button => button.textContent?.includes('Cancel scan')) as HTMLButtonElement
+    await act(async () => { cancel.click(); await Promise.resolve(); await Promise.resolve() })
+    expect(container.querySelector('[role="alert"]')?.textContent).toContain('cancel unavailable')
+    expect(cancel).toBeEnabled()
+    await act(async () => { cancel.click(); await Promise.resolve(); await Promise.resolve() })
+    expect(cancelScan).toHaveBeenCalledTimes(2)
+
+    const notify = Array.from(container.querySelectorAll('button')).find(button => button.textContent?.includes('Test notifications')) as HTMLButtonElement
+    await act(async () => { notify.click(); await Promise.resolve(); await Promise.resolve() })
+    expect(container.querySelector('[role="alert"]')?.textContent).toContain('delivery unavailable')
+    await act(async () => { notify.click(); await Promise.resolve(); await Promise.resolve() })
+    expect(container.querySelector('[role="status"]')?.textContent).toContain('1 destination tested')
+    expect(notificationTest).toHaveBeenCalledTimes(2)
+  })
+
   it('keeps the page usable when jobs fail to load', async () => {
     vi.mocked(listJobs).mockRejectedValue(new Error('database unavailable'))
     await act(async () => {
@@ -287,4 +324,32 @@ describe('dashboard', () => {
     })
     expect(listJobs).toHaveBeenCalledTimes(2)
   })
+
+  it('keeps the last live scan count visible as stale and recovers it on retry', async () => {
+    await renderDashboard()
+    vi.mocked(activeScans).mockRejectedValueOnce(new Error('temporary polling failure'))
+    const recentScansMetric = Array.from(container.querySelectorAll('.stat-card')).find(card => card.querySelector('.stat-label')?.textContent === 'Recent scans')
+
+    await act(async () => {
+      await queryClient.refetchQueries({ queryKey: ['active-scans'], exact: true })
+    })
+    await vi.waitFor(() => expect(recentScansMetric?.querySelector('.stat-detail')?.textContent).toContain('1 in progress · stale'))
+
+    const retry = recentScansMetric?.querySelector('.stat-detail .stat-retry') as HTMLButtonElement
+    expect(retry).toBeEnabled()
+    await act(async () => {
+      retry.click()
+      await Promise.resolve()
+    })
+    await vi.waitFor(() => expect(recentScansMetric?.querySelector('.stat-detail')?.textContent).not.toContain('stale'))
+    expect(recentScansMetric?.querySelector('.stat-detail')?.textContent).toContain('1 in progress')
+  })
+
+  it('explains when no notification destination is configured', async () => {
+    vi.mocked(adminStatus).mockResolvedValue({ ...status, notification_destinations: 0 })
+    await renderDashboard()
+
+    expect(container.querySelector('.page-heading')?.textContent).toContain('Notifications are configured by an administrator.')
+    expect(container.querySelector('.page-heading')?.textContent).not.toContain('notification destination configured')
+  })
 })