Files
silo-server/internal/api/handlers/diagnostics_test.go
f235524365 feat(diagnostics): chunked report upload fallback for proxy body caps (#494)
* feat(diagnostics): chunked report upload fallback for proxy body caps

Diagnostics bundles can be up to max_bundle_bytes (10 MiB default), but a
reverse proxy in front of Silo commonly caps request bodies at nginx's
default client_max_body_size of 1 MiB. Such a proxy answers the single-shot
multipart upload with its own 413 before Silo ever sees the request, so any
report over the cap could never be delivered.

Add a chunked upload fallback under /api/v1/diagnostics/reports/uploads:

- POST   /                      {manifest, bundle_bytes} opens a session
- PUT    /{id}/chunks/{index}   streams one ≤768 KiB chunk (proxy-safe)
- POST   /{id}/complete         ingests the assembled bundle
- DELETE /{id}                  best-effort abandon

The assembled bundle goes through the exact same Ingest path as the
single-shot endpoint, so every content check (manifest contract, archive
sha/bytes/entries, quotas, profile attribution) applies identically.
Sessions reuse internal/uploads (the plugin chunked-upload spool manager)
plus a small owner map for per-user isolation; they spool to disk, expire
after 15 minutes, cap at one per user / 16 global, and complete shares the
existing per-user + global in-flight ingest limiter.

/diagnostics/status now advertises upload_chunk_bytes so clients can detect
support; older servers omit the field and clients treat that as
unsupported. The demo guard's diagnostics prefix gains PUT to cover the
chunk route.

Verified end to end against an OpenResty proxy with a 1m body cap: the
single-shot upload 413s, the same 1.6 MiB bundle uploads in three chunks
and lands as an accepted report; also exercised from the tvOS client's
fallback path in the simulator.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(diagnostics): harden chunked upload sessions per review

- Reserve the per-user slot and global cap atomically in init (a
  reservation map counted with live sessions), so concurrent inits by one
  account can no longer fan out past one session or transiently exceed the
  cap. Creation failures roll the reservation back.
- Move chunk body I/O outside the uploads.Manager mutex: a slow client
  streaming one chunk no longer serializes every other session's chunk
  writes, completes, and cancels. A per-chunk in-flight flag rejects
  duplicate concurrent writes to the same offset (ErrChunkBusy → 409), and
  cancel/expiry defer spool-directory removal to the last finishing
  writer.
- Chunk arrivals refresh the session expiry, making the TTL an idle
  timeout instead of an absolute deadline so a slow-but-progressing upload
  cannot expire mid-transfer.
- Extend the request read deadline on chunk PUTs and both deadlines on
  complete, matching the single-shot handler's slow-uplink handling.
- Keep the session when complete's availability re-check fails
  transiently (status load error → 500): only definitive
  disabled/storage-unavailable answers discard the spool, so a retried
  complete succeeds without re-uploading every chunk.
- Reclaim orphaned spool directories at startup (a restart previously
  stranded the old process's partial uploads forever) and sweep expired
  sessions on a timer instead of only from later init traffic.
- Document that session state is process-local and what that means for
  multi-replica deployments.

Adds concurrency/race tests (go test -race) for atomic admission,
same-chunk write exclusion, expiry refresh, transient-status retry, and
startup reclaim.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(diagnostics): count detached chunk writers and lift chunk PUT write deadline

Second review round:

- A canceled session whose slow chunk writer was still draining held a
  connection and spool disk but vanished from every count, so a
  cancel-and-reinit loop could stack unbounded live writers behind the
  16-session cap. The uploads manager now parks such sessions in a
  detached set (exposed as DetachedWriterSessions) until their last
  writer returns, and diagnostics init counts them in its admission gate.
- Chunk PUTs now extend the write deadline as well as the read deadline:
  on an uplink slow enough to eat the server's 120s WriteTimeout, the
  stored chunk's JSON acknowledgement would otherwise be lost and the
  client would retry an already-accepted chunk.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-27 12:07:37 -04:00

894 lines
30 KiB
Go

package handlers
import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
"log/slog"
"mime/multipart"
"net/http"
"net/http/httptest"
"net/textproto"
"strconv"
"strings"
"sync"
"testing"
"time"
"github.com/go-chi/chi/v5"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
"github.com/Silo-Server/silo-server/internal/auth"
"github.com/Silo-Server/silo-server/internal/diagnostics"
)
func TestDiagnosticsUploadHappyPath(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
rec := httptest.NewRecorder()
handler.HandleUpload(rec, newDiagnosticsUploadRequest(t, []diagnosticsPart{
{name: "manifest", contentType: "application/json", body: []byte(`{"ok":true}`)},
{name: "bundle", contentType: "application/gzip", body: []byte("bundle")},
}, accessClaims()))
if rec.Code != http.StatusCreated {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusCreated, rec.Body.String())
}
var resp diagnostics.IngestResult
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
t.Fatalf("decode response: %v", err)
}
if resp.ShortID != "SILO-ABCDEF123456" {
t.Fatalf("short_id = %q, want SILO-ABCDEF123456", resp.ShortID)
}
if service.ingestCalls != 1 {
t.Fatalf("ingest calls = %d, want 1", service.ingestCalls)
}
}
func TestDiagnosticsUploadManifestTooBig(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
logs := captureDiagnosticsLogs(handler)
rec := httptest.NewRecorder()
handler.HandleUpload(rec, newDiagnosticsUploadRequest(t, []diagnosticsPart{
{name: "manifest", contentType: "application/json", body: bytes.Repeat([]byte("x"), int(diagnostics.MaxManifestBytes)+1)},
{name: "bundle", contentType: "application/gzip", body: []byte("bundle")},
}, accessClaims()))
assertDiagnosticsError(t, rec, http.StatusRequestEntityTooLarge, "too_large")
if service.ingestCalls != 0 {
t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
}
assertDiagnosticsRejectionLog(t, logs, "too_large", 42)
}
func TestDiagnosticsUploadWrongPartOrder(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
logs := captureDiagnosticsLogs(handler)
rec := httptest.NewRecorder()
handler.HandleUpload(rec, newDiagnosticsUploadRequest(t, []diagnosticsPart{
{name: "bundle", contentType: "application/gzip", body: []byte("bundle")},
{name: "manifest", contentType: "application/json", body: []byte(`{"ok":true}`)},
}, accessClaims()))
assertDiagnosticsError(t, rec, http.StatusBadRequest, "invalid_bundle")
if service.ingestCalls != 0 {
t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
}
assertDiagnosticsRejectionLog(t, logs, "invalid_bundle", 42)
}
func TestDiagnosticsUploadWrongFirstPartNotDrained(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
// A large, wrongly-named first part must be rejected without draining its
// body: an invalid upload must not be allowed to stream its whole payload
// (holding the in-flight slot) before it gets a 400.
var body bytes.Buffer
writer := multipart.NewWriter(&body)
header := make(textproto.MIMEHeader)
header.Set("Content-Disposition", `form-data; name="bundle"`)
header.Set("Content-Type", "application/gzip")
w, err := writer.CreatePart(header)
if err != nil {
t.Fatalf("create multipart part: %v", err)
}
if _, err := w.Write(bytes.Repeat([]byte("x"), 1<<20)); err != nil {
t.Fatalf("write multipart part: %v", err)
}
if err := writer.Close(); err != nil {
t.Fatalf("close multipart writer: %v", err)
}
counter := &countingReader{r: &body}
req := httptest.NewRequest(http.MethodPost, "/api/v1/diagnostics/reports", counter)
req.Header.Set("Content-Type", writer.FormDataContentType())
req.Header.Set("Authorization", "Bearer token")
req = req.WithContext(apimw.SetClaims(req.Context(), accessClaims()))
rec := httptest.NewRecorder()
handler.HandleUpload(rec, req)
assertDiagnosticsError(t, rec, http.StatusBadRequest, "invalid_bundle")
if service.ingestCalls != 0 {
t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
}
// Only the small multipart framing/headers should have been read; the 1 MiB
// wrong-first-part body must not have been drained.
if counter.n > 128*1024 {
t.Fatalf("read %d bytes from request body, want the wrong first part rejected without draining (< 128 KiB)", counter.n)
}
}
type countingReader struct {
r io.Reader
n int64
}
func (c *countingReader) Read(p []byte) (int, error) {
n, err := c.r.Read(p)
c.n += int64(n)
return n, err
}
func TestDiagnosticsUploadBodyTooLargeLogsRejection(t *testing.T) {
service := newFakeDiagnosticsService()
service.status.MaxBundleBytes = 1
handler := NewDiagnosticsHandler(service)
logs := captureDiagnosticsLogs(handler)
body := strings.NewReader(strings.Repeat("x\n", int(diagnosticsMultipartOverheadBytes/2)+2))
req := httptest.NewRequest(http.MethodPost, "/api/v1/diagnostics/reports", body)
req.Header.Set("Content-Type", "multipart/form-data; boundary=diagnostics-boundary")
req.Header.Set("Authorization", "Bearer token")
req = req.WithContext(apimw.SetClaims(req.Context(), accessClaims()))
rec := httptest.NewRecorder()
handler.HandleUpload(rec, req)
assertDiagnosticsError(t, rec, http.StatusRequestEntityTooLarge, "too_large")
if service.ingestCalls != 0 {
t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
}
assertDiagnosticsRejectionLog(t, logs, "too_large", 42)
}
func TestDiagnosticsUploadMissingPart(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
rec := httptest.NewRecorder()
handler.HandleUpload(rec, newDiagnosticsUploadRequest(t, []diagnosticsPart{
{name: "manifest", contentType: "application/json", body: []byte(`{"ok":true}`)},
}, accessClaims()))
assertDiagnosticsError(t, rec, http.StatusBadRequest, "invalid_bundle")
if service.ingestCalls != 0 {
t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
}
}
func TestDiagnosticsUploadDisabled(t *testing.T) {
service := newFakeDiagnosticsService()
service.status.Status = diagnostics.StatusDisabled
handler := NewDiagnosticsHandler(service)
rec := httptest.NewRecorder()
handler.HandleUpload(rec, newDiagnosticsUploadRequest(t, validDiagnosticsParts(), accessClaims()))
assertDiagnosticsError(t, rec, http.StatusForbidden, "disabled")
if service.ingestCalls != 0 {
t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
}
}
func TestDiagnosticsUploadStorageUnavailable(t *testing.T) {
service := newFakeDiagnosticsService()
service.status.Status = diagnostics.StatusStorageUnavailable
handler := NewDiagnosticsHandler(service)
rec := httptest.NewRecorder()
handler.HandleUpload(rec, newDiagnosticsUploadRequest(t, validDiagnosticsParts(), accessClaims()))
assertDiagnosticsError(t, rec, http.StatusServiceUnavailable, "storage_unavailable")
if service.ingestCalls != 0 {
t.Fatalf("ingest calls = %d, want 0", service.ingestCalls)
}
}
func TestDiagnosticsUploadQuotaExceededSetsRetryAfter(t *testing.T) {
service := newFakeDiagnosticsService()
service.ingestErr = &diagnostics.QuotaError{Kind: diagnostics.QuotaKindReportsPerDay, Limit: 20}
handler := NewDiagnosticsHandler(service)
rec := httptest.NewRecorder()
handler.HandleUpload(rec, newDiagnosticsUploadRequest(t, validDiagnosticsParts(), accessClaims()))
assertDiagnosticsError(t, rec, http.StatusTooManyRequests, "quota_exceeded")
if got := rec.Header().Get("Retry-After"); got != diagnosticsQuotaRetryAfter {
t.Fatalf("Retry-After = %q, want %q", got, diagnosticsQuotaRetryAfter)
}
}
func TestDiagnosticsUploadRejectsAPIKey(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
logs := captureDiagnosticsLogs(handler)
rec := httptest.NewRecorder()
req := newDiagnosticsUploadRequest(t, validDiagnosticsParts(), &auth.Claims{
UserID: 42,
TokenType: auth.TokenTypeAPIKey,
})
req.Header.Set("Authorization", "Bearer sa_test")
handler.HandleUpload(rec, req)
assertDiagnosticsError(t, rec, http.StatusForbidden, "api_key_not_allowed")
if service.statusCalls != 0 {
t.Fatalf("status calls = %d, want 0", service.statusCalls)
}
assertDiagnosticsRejectionLog(t, logs, "api_key_not_allowed", 42)
}
func TestDiagnosticsUploadBusy(t *testing.T) {
service := newFakeDiagnosticsService()
service.started = make(chan struct{})
service.block = make(chan struct{})
handler := NewDiagnosticsHandler(service)
logs := captureDiagnosticsLogs(handler)
firstReq := newDiagnosticsUploadRequest(t, validDiagnosticsParts(), accessClaims())
firstDone := make(chan *httptest.ResponseRecorder, 1)
go func() {
rec := httptest.NewRecorder()
handler.HandleUpload(rec, firstReq)
firstDone <- rec
}()
<-service.started
rec := httptest.NewRecorder()
handler.HandleUpload(rec, newDiagnosticsUploadRequest(t, validDiagnosticsParts(), accessClaims()))
assertDiagnosticsError(t, rec, http.StatusServiceUnavailable, "busy")
if got := rec.Header().Get("Retry-After"); got != diagnosticsBusyRetryAfter {
t.Fatalf("Retry-After = %q, want %q", got, diagnosticsBusyRetryAfter)
}
assertDiagnosticsRejectionLog(t, logs, "busy", 42)
close(service.block)
first := <-firstDone
if first.Code != http.StatusCreated {
t.Fatalf("first status = %d, want %d; body=%s", first.Code, http.StatusCreated, first.Body.String())
}
}
func TestDiagnosticsAdminListReportsParsesFilters(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
req := httptest.NewRequest(
http.MethodGet,
"/api/v1/admin/diagnostics/reports?user_id=42&platform=ios&report_type=crash&from=2026-07-19T10:00:00Z&to=2026-07-20T10:00:00Z&short_id=abcdef123456&limit=25&cursor=next",
nil,
)
rec := httptest.NewRecorder()
handler.HandleAdminListReports(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
if service.listCalls != 1 {
t.Fatalf("list calls = %d, want 1", service.listCalls)
}
filters := service.listFilters
if filters.UserID == nil || *filters.UserID != 42 {
t.Fatalf("UserID = %v, want 42", filters.UserID)
}
if filters.Platform != "ios" || filters.ReportType != "crash" {
t.Fatalf("filters = %#v, want platform ios report_type crash", filters)
}
if filters.From == nil || !filters.From.Equal(time.Date(2026, 7, 19, 10, 0, 0, 0, time.UTC)) {
t.Fatalf("From = %v, want expected timestamp", filters.From)
}
if filters.To == nil || !filters.To.Equal(time.Date(2026, 7, 20, 10, 0, 0, 0, time.UTC)) {
t.Fatalf("To = %v, want expected timestamp", filters.To)
}
if filters.ShortID != "SILO-ABCDEF123456" || filters.Limit != 25 || filters.Cursor != "next" {
t.Fatalf("filters = %#v, want normalized short id, limit, cursor", filters)
}
}
func TestDiagnosticsAdminDownloadReturnsPresignedURLWhenAvailable(t *testing.T) {
service := newFakeDiagnosticsService()
service.getReport = readyDiagnosticsReport()
service.presignURL = "https://storage.example.test/report"
service.effectiveTTL = 5 * time.Minute
handler := NewDiagnosticsHandler(service)
before := time.Now().UTC()
rec := httptest.NewRecorder()
adminDiagnosticsRouter(handler).ServeHTTP(rec, adminDiagnosticsRequest(http.MethodGet, "/diagnostics/reports/report-1/download", adminClaims()))
after := time.Now().UTC()
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
var resp diagnosticsDownloadURLResponse
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
t.Fatalf("decode response: %v", err)
}
if resp.DownloadURL != service.presignURL {
t.Fatalf("download_url = %q, want %q", resp.DownloadURL, service.presignURL)
}
if service.presignExpiry != service.effectiveTTL {
t.Fatalf("presign expiry = %s, want %s", service.presignExpiry, service.effectiveTTL)
}
if resp.ExpiresAt.Before(before.Add(service.effectiveTTL)) || resp.ExpiresAt.After(after.Add(service.effectiveTTL)) {
t.Fatalf("expires_at = %s, want within effective TTL window", resp.ExpiresAt)
}
if service.openCalls != 0 {
t.Fatalf("open calls = %d, want 0", service.openCalls)
}
}
func TestDiagnosticsAdminDownloadStreamsWhenProxyRequested(t *testing.T) {
service := newFakeDiagnosticsService()
service.getReport = readyDiagnosticsReport()
service.presignURL = "https://storage.example.test/report"
service.openData = []byte("gzipped report")
handler := NewDiagnosticsHandler(service)
rec := httptest.NewRecorder()
adminDiagnosticsRouter(handler).ServeHTTP(rec, adminDiagnosticsRequest(http.MethodGet, "/diagnostics/reports/report-1/download?proxy=1", adminClaims()))
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
if service.presignCalls != 0 {
t.Fatalf("presign calls = %d, want 0", service.presignCalls)
}
if service.openCalls != 1 {
t.Fatalf("open calls = %d, want 1", service.openCalls)
}
if rec.Body.String() != "gzipped report" {
t.Fatalf("body = %q, want streamed data", rec.Body.String())
}
}
func TestDiagnosticsAdminDownloadStreamsWhenPresignUnavailable(t *testing.T) {
service := newFakeDiagnosticsService()
service.getReport = readyDiagnosticsReport()
service.presignErr = errors.New("presign unavailable")
service.openData = []byte("gzipped report")
handler := NewDiagnosticsHandler(service)
rec := httptest.NewRecorder()
adminDiagnosticsRouter(handler).ServeHTTP(rec, adminDiagnosticsRequest(http.MethodGet, "/diagnostics/reports/report-1/download", adminClaims()))
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
if got := rec.Header().Get("Content-Type"); got != diagnostics.ReportDownloadContentType {
t.Fatalf("Content-Type = %q, want %q", got, diagnostics.ReportDownloadContentType)
}
if !strings.Contains(rec.Header().Get("Content-Disposition"), "silo-diagnostics-SILO-ABCDEF123456.tar.gz") {
t.Fatalf("Content-Disposition = %q, want diagnostic filename", rec.Header().Get("Content-Disposition"))
}
if rec.Body.String() != "gzipped report" {
t.Fatalf("body = %q, want streamed data", rec.Body.String())
}
}
func TestDiagnosticsAdminListReportsRejectsMalformedQuery(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/diagnostics/reports?user_id=abc", nil)
rec := httptest.NewRecorder()
handler.HandleAdminListReports(rec, req)
assertDiagnosticsError(t, rec, http.StatusBadRequest, "bad_request")
if service.listCalls != 0 {
t.Fatalf("list calls = %d, want 0", service.listCalls)
}
}
func TestDiagnosticsAdminListReportsRejectsMalformedCursor(t *testing.T) {
service := newFakeDiagnosticsService()
service.listErr = diagnostics.ErrInvalidCursor
handler := NewDiagnosticsHandler(service)
req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/diagnostics/reports?cursor=bad", nil)
rec := httptest.NewRecorder()
handler.HandleAdminListReports(rec, req)
assertDiagnosticsError(t, rec, http.StatusBadRequest, "bad_request")
}
func TestDiagnosticsAdminDeleteEmitsAuditEvent(t *testing.T) {
service := newFakeDiagnosticsService()
service.deleteReport = readyDiagnosticsReport()
var logs bytes.Buffer
handler := NewDiagnosticsHandler(service)
handler.logger = slog.New(slog.NewJSONHandler(&logs, nil))
rec := httptest.NewRecorder()
adminDiagnosticsRouter(handler).ServeHTTP(rec, adminDiagnosticsRequest(http.MethodDelete, "/diagnostics/reports/report-1", adminClaims()))
if rec.Code != http.StatusNoContent {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusNoContent, rec.Body.String())
}
logLine := logs.String()
if !strings.Contains(logLine, `"msg":"diagnostic report deleted"`) ||
!strings.Contains(logLine, `"admin_user_id":7`) ||
!strings.Contains(logLine, `"report_id":"report-1"`) {
t.Fatalf("audit log = %s", logLine)
}
}
func TestDiagnosticsAdminRoutesRejectNonActingAdmin(t *testing.T) {
service := newFakeDiagnosticsService()
handler := NewDiagnosticsHandler(service)
router := chi.NewRouter()
router.Use(apimw.RequireActingAdmin(nil))
RegisterAdminDiagnosticsRoutes(router, handler)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, adminDiagnosticsRequest(http.MethodGet, "/diagnostics/reports", &auth.Claims{
UserID: 8,
Role: "user",
TokenType: auth.TokenTypeAccess,
}))
if rec.Code != http.StatusForbidden {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusForbidden, rec.Body.String())
}
if service.listCalls != 0 {
t.Fatalf("list calls = %d, want 0", service.listCalls)
}
}
func TestDiagnosticsUploadsEnablementRequiresStorageProbe(t *testing.T) {
settings := &fakeServerSettingsStore{values: map[string]string{}}
handler := &AdminHandler{SettingsRepo: settings}
router := chi.NewRouter()
router.Put("/admin/settings/{key}", handler.HandleUpdateSetting)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, httptest.NewRequest(
http.MethodPut,
"/admin/settings/"+diagnostics.KeyUploadsEnabled,
strings.NewReader(`{"value":"true"}`),
))
assertDiagnosticsError(t, rec, http.StatusBadRequest, "storage_unavailable")
if settings.values[diagnostics.KeyUploadsEnabled] != "" {
t.Fatalf("setting persisted = %q, want empty", settings.values[diagnostics.KeyUploadsEnabled])
}
store := &fakeDiagnosticsEnablementStore{bucket: "private"}
handler.DiagnosticsStore = store
rec = httptest.NewRecorder()
router.ServeHTTP(rec, httptest.NewRequest(
http.MethodPut,
"/admin/settings/"+diagnostics.KeyUploadsEnabled,
strings.NewReader(`{"value":"true"}`),
))
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
if settings.values[diagnostics.KeyUploadsEnabled] != "true" {
t.Fatalf("setting persisted = %q, want true", settings.values[diagnostics.KeyUploadsEnabled])
}
if !sameStringSlice(store.ops, []string{"put:diagnostics/.probe", "delete:diagnostics/.probe"}) {
t.Fatalf("probe ops = %v, want put/delete probe", store.ops)
}
}
func TestDiagnosticsNumericSettingsAcceptBoundaries(t *testing.T) {
cases := []struct {
name string
key string
value string
initial map[string]string
}{
{name: "bundle minimum", key: diagnostics.KeyMaxBundleBytes, value: "1048576", initial: map[string]string{diagnostics.KeyMaxUncompressedBytes: "1073741824"}},
{name: "bundle maximum", key: diagnostics.KeyMaxBundleBytes, value: "268435456", initial: map[string]string{diagnostics.KeyMaxUncompressedBytes: "1073741824", diagnostics.KeyMaxBytesPerUser: "10737418240"}},
{name: "uncompressed bundle floor", key: diagnostics.KeyMaxUncompressedBytes, value: "1048576", initial: map[string]string{diagnostics.KeyMaxBundleBytes: "1048576"}},
{name: "uncompressed maximum", key: diagnostics.KeyMaxUncompressedBytes, value: "1073741824", initial: map[string]string{diagnostics.KeyMaxBundleBytes: "1048576"}},
{name: "reports minimum", key: diagnostics.KeyMaxReportsPerUserDay, value: "1"},
{name: "reports maximum", key: diagnostics.KeyMaxReportsPerUserDay, value: "1000"},
{name: "retention minimum", key: diagnostics.KeyRetentionDays, value: "1"},
{name: "retention maximum", key: diagnostics.KeyRetentionDays, value: "365"},
{name: "user bytes minimum", key: diagnostics.KeyMaxBytesPerUser, value: "10485760"},
{name: "user bytes maximum", key: diagnostics.KeyMaxBytesPerUser, value: "10737418240"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
settings := &fakeServerSettingsStore{values: tc.initial}
rec := updateDiagnosticsSetting(t, &AdminHandler{SettingsRepo: settings}, tc.key, tc.value)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
if settings.values[tc.key] != tc.value {
t.Fatalf("stored value = %q, want %q", settings.values[tc.key], tc.value)
}
})
}
}
func TestDiagnosticsNumericSettingsRejectOutOfRangeValues(t *testing.T) {
cases := []struct {
name string
key string
value string
initial map[string]string
}{
{name: "bundle below minimum", key: diagnostics.KeyMaxBundleBytes, value: "1048575"},
{name: "bundle above maximum", key: diagnostics.KeyMaxBundleBytes, value: "268435457", initial: map[string]string{diagnostics.KeyMaxUncompressedBytes: "1073741824"}},
{name: "bundle above uncompressed", key: diagnostics.KeyMaxBundleBytes, value: "67108865", initial: map[string]string{diagnostics.KeyMaxUncompressedBytes: "67108864"}},
{name: "uncompressed below bundle", key: diagnostics.KeyMaxUncompressedBytes, value: "10485759", initial: map[string]string{diagnostics.KeyMaxBundleBytes: "10485760"}},
{name: "bundle above per-user cap", key: diagnostics.KeyMaxBundleBytes, value: "104857600", initial: map[string]string{diagnostics.KeyMaxUncompressedBytes: "1073741824", diagnostics.KeyMaxBytesPerUser: "10485760"}},
{name: "per-user below bundle", key: diagnostics.KeyMaxBytesPerUser, value: "10485760", initial: map[string]string{diagnostics.KeyMaxBundleBytes: "104857600", diagnostics.KeyMaxUncompressedBytes: "1073741824"}},
{name: "uncompressed above maximum", key: diagnostics.KeyMaxUncompressedBytes, value: "1073741825"},
{name: "reports below minimum", key: diagnostics.KeyMaxReportsPerUserDay, value: "0"},
{name: "reports above maximum", key: diagnostics.KeyMaxReportsPerUserDay, value: "1001"},
{name: "retention below minimum", key: diagnostics.KeyRetentionDays, value: "0"},
{name: "retention above maximum", key: diagnostics.KeyRetentionDays, value: "366"},
{name: "user bytes below minimum", key: diagnostics.KeyMaxBytesPerUser, value: "10485759"},
{name: "user bytes above maximum", key: diagnostics.KeyMaxBytesPerUser, value: "10737418241"},
{name: "not an integer", key: diagnostics.KeyRetentionDays, value: "thirty"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
settings := &fakeServerSettingsStore{values: tc.initial}
rec := updateDiagnosticsSetting(t, &AdminHandler{SettingsRepo: settings}, tc.key, tc.value)
body := rec.Body.String()
assertDiagnosticsError(t, rec, http.StatusBadRequest, "bad_request")
if _, stored := settings.values[tc.key]; stored {
t.Fatalf("invalid value was stored: %#v", settings.values)
}
if !strings.Contains(body, tc.key) {
t.Fatalf("error body = %s, want clear error naming %s", body, tc.key)
}
})
}
}
type diagnosticsPart struct {
name string
contentType string
body []byte
}
func validDiagnosticsParts() []diagnosticsPart {
return []diagnosticsPart{
{name: "manifest", contentType: "application/json", body: []byte(`{"ok":true}`)},
{name: "bundle", contentType: "application/gzip", body: []byte("bundle")},
}
}
func newDiagnosticsUploadRequest(t *testing.T, parts []diagnosticsPart, claims *auth.Claims) *http.Request {
t.Helper()
var body bytes.Buffer
writer := multipart.NewWriter(&body)
for _, part := range parts {
header := make(textproto.MIMEHeader)
header.Set("Content-Disposition", `form-data; name="`+part.name+`"`)
header.Set("Content-Type", part.contentType)
w, err := writer.CreatePart(header)
if err != nil {
t.Fatalf("create multipart part: %v", err)
}
if _, err := w.Write(part.body); err != nil {
t.Fatalf("write multipart part: %v", err)
}
}
if err := writer.Close(); err != nil {
t.Fatalf("close multipart writer: %v", err)
}
req := httptest.NewRequest(http.MethodPost, "/api/v1/diagnostics/reports", &body)
req.Header.Set("Content-Type", writer.FormDataContentType())
req.Header.Set("Authorization", "Bearer token")
if claims != nil {
req = req.WithContext(apimw.SetClaims(req.Context(), claims))
}
return req
}
func accessClaims() *auth.Claims {
return &auth.Claims{
UserID: 42,
TokenType: auth.TokenTypeAccess,
}
}
func assertDiagnosticsError(t *testing.T, rec *httptest.ResponseRecorder, status int, code string) {
t.Helper()
if rec.Code != status {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, status, rec.Body.String())
}
var resp struct {
Error string `json:"error"`
}
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
t.Fatalf("decode error response: %v", err)
}
if resp.Error != code {
t.Fatalf("error = %q, want %q", resp.Error, code)
}
}
func captureDiagnosticsLogs(handler *DiagnosticsHandler) *bytes.Buffer {
var logs bytes.Buffer
handler.logger = slog.New(slog.NewJSONHandler(&logs, nil))
return &logs
}
func assertDiagnosticsRejectionLog(t *testing.T, logs *bytes.Buffer, reason string, userID int) {
t.Helper()
line := logs.String()
for _, want := range []string{
`"msg":"diagnostic report rejected"`,
`"component":"diagnostics"`,
`"result":"rejected"`,
`"reason":"` + reason + `"`,
`"user_id":` + strconv.Itoa(userID),
} {
if !strings.Contains(line, want) {
t.Fatalf("rejection log = %s, want %s", line, want)
}
}
}
func updateDiagnosticsSetting(t *testing.T, handler *AdminHandler, key, value string) *httptest.ResponseRecorder {
t.Helper()
router := chi.NewRouter()
router.Put("/admin/settings/{key}", handler.HandleUpdateSetting)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, httptest.NewRequest(
http.MethodPut,
"/admin/settings/"+key,
strings.NewReader(`{"value":"`+value+`"}`),
))
return rec
}
func readyDiagnosticsReport() *diagnostics.Report {
blobBucket := "private"
blobKey := "diagnostics/7/report-1.tar.gz"
blobBytes := int64(13)
return &diagnostics.Report{
ID: "report-1",
ShortID: "SILO-ABCDEF123456",
UserID: 7,
State: diagnostics.StateReady,
ReportType: "crash",
Platform: "android-tv",
BlobBucket: &blobBucket,
BlobKey: &blobKey,
BlobBytes: &blobBytes,
}
}
func adminDiagnosticsRouter(handler *DiagnosticsHandler) chi.Router {
router := chi.NewRouter()
RegisterAdminDiagnosticsRoutes(router, handler)
return router
}
func adminDiagnosticsRequest(method, target string, claims *auth.Claims) *http.Request {
req := httptest.NewRequest(method, target, nil)
if claims != nil {
req = req.WithContext(apimw.SetClaims(req.Context(), claims))
}
return req
}
func adminClaims() *auth.Claims {
return &auth.Claims{
UserID: 7,
Role: "admin",
TokenType: auth.TokenTypeAccess,
}
}
type fakeDiagnosticsService struct {
mu sync.Mutex
status diagnostics.Status
statusErr error
statusCalls int
ingestCalls int
ingestErr error
lastManifest []byte
lastBundleBytes int64
started chan struct{}
block chan struct{}
listCalls int
listFilters diagnostics.ListFilters
listResult diagnostics.ListResult
listErr error
getReport *diagnostics.Report
getErr error
presignURL string
presignErr error
presignCalls int
presignExpiry time.Duration
effectiveTTL time.Duration
openCalls int
openData []byte
openErr error
deleteReport *diagnostics.Report
deleteErr error
}
func newFakeDiagnosticsService() *fakeDiagnosticsService {
return &fakeDiagnosticsService{
status: diagnostics.Status{
Status: diagnostics.StatusAvailable,
ServerInstanceID: "server-1",
AcceptedSchemaVersions: []int{1},
MaxBundleBytes: diagnostics.DefaultMaxBundleBytes,
MaxManifestBytes: diagnostics.MaxManifestBytes,
RetentionDays: diagnostics.DefaultRetentionDays,
ConsentNoticeVersion: diagnostics.DefaultConsentNoticeVer,
},
}
}
func (f *fakeDiagnosticsService) Status(context.Context, int) (diagnostics.Status, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.statusCalls++
if f.statusErr != nil {
return diagnostics.Status{}, f.statusErr
}
return f.status, nil
}
func (f *fakeDiagnosticsService) setStatusErr(err error) {
f.mu.Lock()
f.statusErr = err
f.mu.Unlock()
}
func (f *fakeDiagnosticsService) Ingest(_ context.Context, _ int, _ *string, manifestJSON []byte, bundle io.Reader) (diagnostics.IngestResult, error) {
f.mu.Lock()
f.ingestCalls++
f.lastManifest = append([]byte(nil), manifestJSON...)
started := f.started
block := f.block
err := f.ingestErr
f.mu.Unlock()
if started != nil {
close(started)
}
if block != nil {
<-block
}
if err != nil {
return diagnostics.IngestResult{}, err
}
consumed, err := io.Copy(io.Discard, bundle)
if err != nil {
return diagnostics.IngestResult{}, err
}
f.mu.Lock()
f.lastBundleBytes = consumed
f.mu.Unlock()
return diagnostics.IngestResult{
ReportID: "11111111-1111-1111-1111-111111111111",
ShortID: "SILO-ABCDEF123456",
}, nil
}
func (f *fakeDiagnosticsService) ListForAdmin(_ context.Context, filters diagnostics.ListFilters) (diagnostics.ListResult, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.listCalls++
f.listFilters = filters
if f.listErr != nil {
return diagnostics.ListResult{}, f.listErr
}
return f.listResult, nil
}
func (f *fakeDiagnosticsService) GetReport(context.Context, string) (*diagnostics.Report, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.getErr != nil {
return nil, f.getErr
}
if f.getReport != nil {
return f.getReport, nil
}
return readyDiagnosticsReport(), nil
}
func (f *fakeDiagnosticsService) PresignReportDownload(_ context.Context, _ *diagnostics.Report, expiry time.Duration) (string, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.presignCalls++
f.presignExpiry = expiry
if f.presignErr != nil {
return "", f.presignErr
}
return f.presignURL, nil
}
func (f *fakeDiagnosticsService) EffectiveReportDownloadTTL(requested time.Duration) time.Duration {
f.mu.Lock()
defer f.mu.Unlock()
if f.effectiveTTL > 0 {
return f.effectiveTTL
}
return requested
}
func (f *fakeDiagnosticsService) OpenReportDownload(context.Context, *diagnostics.Report) (io.ReadCloser, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.openCalls++
if f.openErr != nil {
return nil, f.openErr
}
return io.NopCloser(bytes.NewReader(f.openData)), nil
}
func (f *fakeDiagnosticsService) DeleteReport(context.Context, string) (*diagnostics.Report, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.deleteErr != nil {
return nil, f.deleteErr
}
if f.deleteReport != nil {
return f.deleteReport, nil
}
return readyDiagnosticsReport(), nil
}
type fakeDiagnosticsEnablementStore struct {
bucket string
ops []string
}
func (f *fakeDiagnosticsEnablementStore) PutStream(_ context.Context, _ string, key string, r io.Reader, _ string) error {
if _, err := io.ReadAll(r); err != nil {
return err
}
f.ops = append(f.ops, "put:"+key)
return nil
}
func (f *fakeDiagnosticsEnablementStore) DeleteObject(_ context.Context, _ string, key string) error {
f.ops = append(f.ops, "delete:"+key)
return nil
}
func (f *fakeDiagnosticsEnablementStore) Bucket() string {
return f.bucket
}
func sameStringSlice(a, b []string) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}