Merge pull request #450 from Silo-Server/feat/client-diagnostics-server

fix(diagnostics): post-merge review follow-ups for #445

internal/api/handlers/diagnostics.go

@@ -220,12 +220,16 @@ func nextDiagnosticsPart(mr *multipart.Reader, expectedName, expectedContentType
 	if err != nil {
 		return nil, err
 	}
+	// Reject a mismatched part without calling part.Close(): Close drains the
+	// unread part body first, so a max-size wrongly-named/typed first part would
+	// stream up to the bundle limit — holding the per-user/global in-flight slot —
+	// before we return 400. Abandoning the part returns immediately; net/http then
+	// discards only a small bounded prefix of the request before closing the
+	// connection, so malformed uploads fail promptly under load.
 	if part.FormName() != expectedName {
-		_ = part.Close()
 		return nil, errDiagnosticsUnexpectedPart
 	}
 	if !diagnosticsContentTypeMatches(part.Header.Get("Content-Type"), expectedContentType) {
-		_ = part.Close()
 		return nil, errDiagnosticsUnexpectedPart
 	}
 	return part, nil

internal/api/handlers/diagnostics_test.go

@@ -85,6 +85,60 @@ func TestDiagnosticsUploadWrongPartOrder(t *testing.T) {
 	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

internal/diagnostics/repo.go

@@ -297,7 +297,8 @@ func (r *PostgresRepository) DeleteByID(ctx context.Context, id string) (*Report
 		WHERE id = $1::uuid
 		RETURNING id::text, short_id, user_id, profile_id, state, captured_at, received_at,
 		          report_type, platform, app_version, crash_summary, manifest, playback_session_ids,
-		          blob_bucket, blob_key, blob_bytes, uncompressed_bytes, blob_sha256
+		          blob_bucket, blob_key, blob_bytes, uncompressed_bytes, blob_sha256,
+		          COALESCE(manifest->'report'->>'app_build', '')
 	`, id)
 	report, err := scanReport(row)
 	if err != nil {
@@ -314,13 +315,13 @@ func (r *PostgresRepository) RetentionCandidates(ctx context.Context, olderThan
 		return nil, nil
 	}
 
-	query := reportListSelectSQL() + `
+	query := reportCleanupSelectSQL() + `
 		WHERE received_at < $1
 		ORDER BY received_at ASC, id ASC
 	`
 	args := []any{olderThan}
 	if olderThan.IsZero() {
-		query = reportListSelectSQL() + `
+		query = reportCleanupSelectSQL() + `
 			WHERE false
 		`
 		args = nil
@@ -341,12 +342,12 @@ func (r *PostgresRepository) RetentionCandidates(ctx context.Context, olderThan
 				) ranked
 				WHERE retained_bytes > $1
 			)
-		` + reportListSelectSQL() + `
+		` + reportCleanupSelectSQL() + `
 			WHERE id IN (SELECT id FROM quota_candidates)
 			ORDER BY received_at ASC, id ASC
 		`
 		if olderThan.IsZero() {
-			return r.queryReportSummaries(ctx, byteCapQuery, perUserByteCap)
+			return r.queryReportCleanup(ctx, byteCapQuery, perUserByteCap)
 		}
 
 		query = `
@@ -363,14 +364,14 @@ func (r *PostgresRepository) RetentionCandidates(ctx context.Context, olderThan
 				) ranked
 				WHERE retained_bytes > $2
 			)
-		` + reportListSelectSQL() + `
+		` + reportCleanupSelectSQL() + `
 			WHERE received_at < $1 OR id IN (SELECT id FROM quota_candidates)
 			ORDER BY received_at ASC, id ASC
 		`
 		args = []any{olderThan, perUserByteCap}
 	}
 
-	return r.queryReportSummaries(ctx, query, args...)
+	return r.queryReportCleanup(ctx, query, args...)
 }
 
 func (r *PostgresRepository) StaleReceiving(ctx context.Context, grace time.Duration) ([]Report, error) {
@@ -378,7 +379,7 @@ func (r *PostgresRepository) StaleReceiving(ctx context.Context, grace time.Dura
 		grace = time.Hour
 	}
 	cutoff := time.Now().UTC().Add(-grace)
-	return r.queryReportSummaries(ctx, reportListSelectSQL()+`
+	return r.queryReportCleanup(ctx, reportCleanupSelectSQL()+`
 		WHERE state IN ('receiving', 'failed')
 		  AND received_at < $1
 		ORDER BY received_at ASC, id ASC
@@ -417,9 +418,10 @@ func (r *PostgresRepository) LiveBlobKeys(ctx context.Context, keys []string) (m
 	return live, nil
 }
 
-// queryReportSummaries runs a reportListSelectSQL-shaped query; the returned
-// reports carry no manifest. Only list/cleanup paths use it.
-func (r *PostgresRepository) queryReportSummaries(ctx context.Context, query string, args ...any) ([]Report, error) {
+// queryReportCleanup runs a reportCleanupSelectSQL-shaped query; the returned
+// reports carry no manifest and no app_build. Only the retention and
+// stale-cleanup batches use it.
+func (r *PostgresRepository) queryReportCleanup(ctx context.Context, query string, args ...any) ([]Report, error) {
 	rows, err := r.pool.Query(ctx, query, args...)
 	if err != nil {
 		return nil, fmt.Errorf("query diagnostic reports: %w", err)
@@ -428,7 +430,7 @@ func (r *PostgresRepository) queryReportSummaries(ctx context.Context, query str
 
 	reports := []Report{}
 	for rows.Next() {
-		report, err := scanReportSummary(rows)
+		report, err := scanReportCleanup(rows)
 		if err != nil {
 			return nil, err
 		}
@@ -467,19 +469,38 @@ func validateInsertReceivingInput(input InsertReceivingInput) error {
 
 // reportSelectSQL is the full projection including the manifest JSONB (up to
 // MaxManifestBytes per row). Reserve it for single-report detail/download paths;
-// list and cleanup queries use reportListSelectSQL so browsing a page or running
-// a retention/stale batch doesn't drag every report's manifest through the DB.
+// the admin list uses reportListSelectSQL and retention/stale batches use
+// reportCleanupSelectSQL so browsing a page or running a cleanup pass doesn't
+// drag every report's manifest through the DB.
 func reportSelectSQL() string {
 	return `SELECT id::text, short_id, user_id, profile_id, state, captured_at, received_at,
 		       report_type, platform, app_version, crash_summary, manifest, playback_session_ids,
-		       blob_bucket, blob_key, blob_bytes, uncompressed_bytes, blob_sha256
+		       blob_bucket, blob_key, blob_bytes, uncompressed_bytes, blob_sha256,
+		       COALESCE(manifest->'report'->>'app_build', '')
 		FROM client_diagnostic_reports`
 }
 
-// reportListSelectSQL mirrors reportSelectSQL but omits the manifest column.
-// Reports scanned with scanReportSummary therefore have a nil Manifest; the
-// list and cleanup callers only need summary and blob fields.
+// reportListSelectSQL is the admin-list projection: it omits the full manifest
+// column but still extracts app_build from the manifest JSONB (a cheap text
+// extraction) so list rows can show the build number without shipping the whole
+// manifest. Reports scanned with scanReportSummary therefore have a nil
+// Manifest. Cleanup batches must not pay for the JSONB extraction, so they use
+// reportCleanupSelectSQL instead.
 func reportListSelectSQL() string {
+	return `SELECT id::text, short_id, user_id, profile_id, state, captured_at, received_at,
+		       report_type, platform, app_version, crash_summary, playback_session_ids,
+		       blob_bucket, blob_key, blob_bytes, uncompressed_bytes, blob_sha256,
+		       COALESCE(manifest->'report'->>'app_build', '')
+		FROM client_diagnostic_reports`
+}
+
+// reportCleanupSelectSQL is the retention/stale-cleanup projection. It omits
+// both the manifest column and the app_build extraction because cleanup only
+// needs id/state/blob fields to delete a row and its object; touching the JSONB
+// per candidate would turn a scheduled pass over many large manifests into
+// needless manifest I/O. Reports scanned with scanReportCleanup therefore have
+// a nil Manifest and an empty AppBuild.
+func reportCleanupSelectSQL() string {
 	return `SELECT id::text, short_id, user_id, profile_id, state, captured_at, received_at,
 		       report_type, platform, app_version, crash_summary, playback_session_ids,
 		       blob_bucket, blob_key, blob_bytes, uncompressed_bytes, blob_sha256
@@ -553,6 +574,7 @@ func scanReport(row reportScanner) (*Report, error) {
 		&nulls.blobBytes,
 		&nulls.uncompressedBytes,
 		&nulls.blobSHA256,
+		&report.AppBuild,
 	); err != nil {
 		return nil, fmt.Errorf("scan diagnostic report: %w", err)
 	}
@@ -563,12 +585,12 @@ func scanReport(row reportScanner) (*Report, error) {
 	return &report, nil
 }
 
-// scanReportSummary scans the reportListSelectSQL projection, which omits the
-// manifest; the returned report's Manifest is left nil.
-func scanReportSummary(row reportScanner) (*Report, error) {
-	var report Report
-	var nulls reportNulls
-	if err := row.Scan(
+// summaryScanTargets returns the scan destinations shared by the admin-list and
+// cleanup projections, in column order. The two scanners differ only in whether
+// app_build is appended, so keeping the common prefix in one place stops the
+// long Scan lists from drifting apart.
+func summaryScanTargets(report *Report, nulls *reportNulls) []any {
+	return []any{
 		&report.ID,
 		&report.ShortID,
 		&report.UserID,
@@ -586,7 +608,29 @@ func scanReportSummary(row reportScanner) (*Report, error) {
 		&nulls.blobBytes,
 		&nulls.uncompressedBytes,
 		&nulls.blobSHA256,
-	); err != nil {
+	}
+}
+
+// scanReportSummary scans the reportListSelectSQL projection, which omits the
+// manifest but includes app_build; the returned report's Manifest is left nil.
+func scanReportSummary(row reportScanner) (*Report, error) {
+	var report Report
+	var nulls reportNulls
+	targets := append(summaryScanTargets(&report, &nulls), &report.AppBuild)
+	if err := row.Scan(targets...); err != nil {
+		return nil, fmt.Errorf("scan diagnostic report: %w", err)
+	}
+	nulls.apply(&report)
+	return &report, nil
+}
+
+// scanReportCleanup scans the reportCleanupSelectSQL projection, which omits
+// both the manifest and app_build; the returned report's Manifest is nil and
+// AppBuild is empty. Cleanup only reads id/state/blob fields.
+func scanReportCleanup(row reportScanner) (*Report, error) {
+	var report Report
+	var nulls reportNulls
+	if err := row.Scan(summaryScanTargets(&report, &nulls)...); err != nil {
 		return nil, fmt.Errorf("scan diagnostic report: %w", err)
 	}
 	nulls.apply(&report)

internal/diagnostics/service.go

@@ -1,6 +1,7 @@
 package diagnostics
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -181,6 +182,17 @@ func (s *Service) Ingest(ctx context.Context, userID int, profileID *string, man
 		return IngestResult{}, err
 	}
 	crashSummary := crashSummaryFromManifest(manifest)
+	// ExpectedBlobBytes reserves the client-claimed archive size against the
+	// per-user byte quota before the bundle streams in. Trusting the claim here is
+	// safe: a report only reaches MarkReady (which rewrites blob_bytes to the
+	// measured size) after archiveMatches confirms the claimed bytes/sha/entries
+	// exactly equal the streamed bundle, so no stored report can ever occupy more
+	// bytes than it reserved. A client that under-claims to under-reserve then
+	// uploads a larger bundle fails archiveMatches and is compensated (blob
+	// deleted, row marked failed) before it counts, gaining no storage. The
+	// residual cost of such rejected churn is bandwidth, not storage — bounded by
+	// the per-account single in-flight upload cap, not by max_bytes_per_user,
+	// which is a storage cap enforced oldest-first by retention.
 	reserved, err := s.reports.InsertReceiving(ctx, InsertReceivingInput{
 		UserID:               userID,
 		ProfileID:            reportProfileID,
@@ -536,9 +548,25 @@ func embeddedManifestMatches(received, embedded []byte) bool {
 }
 
 func decodeJSONObject(data []byte) (map[string]any, error) {
+	dec := json.NewDecoder(bytes.NewReader(data))
+	// UseNumber keeps every JSON number as its exact decimal text (json.Number)
+	// instead of float64. Without it, two manifests that differ only in a large
+	// integer above 2^53 (e.g. log_summary.lines) both round to the same float
+	// and DeepEqual reports a false match, letting a tampered bundle through.
+	dec.UseNumber()
 	var obj map[string]any
-	if err := json.Unmarshal(data, &obj); err != nil {
+	if err := dec.Decode(&obj); err != nil {
 		return nil, err
 	}
+	// json.Unmarshal rejects trailing bytes after the value; Decoder.Decode does
+	// not, and Decoder.More() misses a stray closing delimiter (e.g. `{...}}`)
+	// because it only reports iteration inside an array/object. Re-assert that
+	// strictness by requiring the stream to be exhausted: a second Decode must
+	// hit io.EOF, so a manifest with any trailing token — junk or an extra
+	// `}`/`]` — is rejected and cannot compare equal. Trailing whitespace is
+	// fine; the decoder skips it and still returns io.EOF.
+	if err := dec.Decode(new(json.RawMessage)); err != io.EOF {
+		return nil, errors.New("unexpected trailing data after JSON object")
+	}
 	return obj, nil
 }

internal/diagnostics/service_test.go

@@ -160,6 +160,48 @@ func TestServiceIngestRejectsEmbeddedManifestMismatch(t *testing.T) {
 	}
 }
 
+func TestEmbeddedManifestMatchesPreservesLargeIntegers(t *testing.T) {
+	// The received (part-1) and embedded manifests differ only in
+	// log_summary.lines by 1, at a magnitude above 2^53 where float64 cannot
+	// represent adjacent integers. Decoding with UseNumber must keep them distinct
+	// so a bundle whose embedded manifest was tampered is rejected, not accepted.
+	received := []byte(`{"log_summary":{"lines":9007199254740993},"archive":{"bytes":1}}`)
+	embedded := []byte(`{"log_summary":{"lines":9007199254740992}}`)
+	if embeddedManifestMatches(received, embedded) {
+		t.Fatal("embeddedManifestMatches = true for manifests differing by a large integer, want false")
+	}
+
+	// Sanity: with the archive object stripped from the received manifest, an
+	// otherwise identical large integer still compares equal.
+	sameReceived := []byte(`{"log_summary":{"lines":9007199254740993},"archive":{"bytes":1}}`)
+	sameEmbedded := []byte(`{"log_summary":{"lines":9007199254740993}}`)
+	if !embeddedManifestMatches(sameReceived, sameEmbedded) {
+		t.Fatal("embeddedManifestMatches = false for identical manifests, want true")
+	}
+}
+
+func TestEmbeddedManifestMatchesRejectsTrailingData(t *testing.T) {
+	// A stray closing delimiter after an otherwise-matching object must be
+	// rejected. Decoder.More() returns false here (it only tracks array/object
+	// iteration), so the decode-to-EOF check is what catches it; without it a
+	// bundle whose embedded manifest is not a single clean JSON object would
+	// still compare equal to the accepted part-1 manifest.
+	received := []byte(`{"log_summary":{"lines":1},"archive":{"bytes":1}}`)
+	for _, trailer := range []string{"}", "]", "garbage", `{"extra":1}`} {
+		embedded := []byte(`{"log_summary":{"lines":1}}` + trailer)
+		if embeddedManifestMatches(received, embedded) {
+			t.Fatalf("embeddedManifestMatches = true for embedded manifest with trailing %q, want false", trailer)
+		}
+	}
+
+	// Trailing whitespace after the object is not trailing data; it must still
+	// compare equal.
+	embeddedWithWhitespace := []byte("{\"log_summary\":{\"lines\":1}}\n  ")
+	if !embeddedManifestMatches(received, embeddedWithWhitespace) {
+		t.Fatal("embeddedManifestMatches = false for embedded manifest with trailing whitespace, want true")
+	}
+}
+
 func TestServiceIngestReturnsStorageErrorForMidStreamPutFailure(t *testing.T) {
 	bundle, info := testDiagnosticsBundle(t)
 	putErr := errors.New("s3: connection reset by peer")

internal/diagnostics/types.go

@@ -71,16 +71,20 @@ type BlobInfo struct {
 }
 
 type Report struct {
-	ID                 string          `json:"id"`
-	ShortID            string          `json:"short_id"`
-	UserID             int             `json:"user_id"`
-	ProfileID          *string         `json:"profile_id,omitempty"`
-	State              ReportState     `json:"state"`
-	CapturedAt         time.Time       `json:"captured_at"`
-	ReceivedAt         time.Time       `json:"received_at"`
-	ReportType         string          `json:"report_type"`
-	Platform           string          `json:"platform"`
-	AppVersion         string          `json:"app_version"`
+	ID         string      `json:"id"`
+	ShortID    string      `json:"short_id"`
+	UserID     int         `json:"user_id"`
+	ProfileID  *string     `json:"profile_id,omitempty"`
+	State      ReportState `json:"state"`
+	CapturedAt time.Time   `json:"captured_at"`
+	ReceivedAt time.Time   `json:"received_at"`
+	ReportType string      `json:"report_type"`
+	Platform   string      `json:"platform"`
+	AppVersion string      `json:"app_version"`
+	// AppBuild is projected from manifest->'report'->>'app_build' so list rows
+	// (which omit the full manifest) can still show the build number without
+	// dragging every report's manifest JSONB through the DB. Empty when absent.
+	AppBuild           string          `json:"app_build"`
 	CrashSummary       *string         `json:"crash_summary,omitempty"`
 	Manifest           json.RawMessage `json:"manifest,omitempty"`
 	PlaybackSessionIDs []string        `json:"playback_session_ids"`

web/src/api/types.ts

@@ -2554,7 +2554,10 @@ export interface ClientDiagnosticManifest {
   [key: string]: unknown;
 }
 
-export interface DiagnosticReport {
+// DiagnosticReportSummary is the shape returned by the admin list endpoint,
+// which omits the full manifest JSONB for cheap paging. `app_build` is projected
+// out of the manifest server-side so rows can still show the build number.
+export interface DiagnosticReportSummary {
   id: string;
   short_id: string;
   user_id: number;
@@ -2565,8 +2568,8 @@ export interface DiagnosticReport {
   report_type: DiagnosticReportType;
   platform: DiagnosticPlatform;
   app_version: string;
+  app_build: string;
   crash_summary?: string;
-  manifest: ClientDiagnosticManifest;
   playback_session_ids: string[];
   blob_bucket?: string;
   blob_key?: string;
@@ -2575,8 +2578,14 @@ export interface DiagnosticReport {
   blob_sha256?: string;
 }
 
+// DiagnosticReport is the full detail shape (GetByID), which additionally
+// includes the parsed manifest for the detail pane.
+export interface DiagnosticReport extends DiagnosticReportSummary {
+  manifest: ClientDiagnosticManifest;
+}
+
 export interface DiagnosticReportListResponse {
-  reports: DiagnosticReport[];
+  reports: DiagnosticReportSummary[];
   next_cursor?: string;
 }
 

web/src/hooks/queries/admin/diagnostics.ts

@@ -5,6 +5,7 @@ import { api, apiResponse } from "@/api/client";
 import type {
   DiagnosticReport,
   DiagnosticReportListResponse,
+  DiagnosticReportSummary,
   DiagnosticStatus,
 } from "@/api/types";
 import { adminKeys } from "@/hooks/queries/keys";
@@ -75,7 +76,7 @@ export function useDeleteDiagnosticReport() {
   });
 }
 
-export async function downloadDiagnosticReport(report: DiagnosticReport) {
+export async function downloadDiagnosticReport(report: DiagnosticReportSummary) {
   // Always stream the bundle through the server (proxy mode) instead of
   // following a presigned URL. When S3Private points at an endpoint only the
   // server can reach (an internal MinIO/R2 gateway), a presigned URL sends the

web/src/pages/AdminDiagnostics.tsx

@@ -4,7 +4,7 @@ import { Link, useSearchParams } from "react-router";
 import { Bug, Download, ExternalLink, FilterX, Trash2, TriangleAlert } from "lucide-react";
 import { toast } from "sonner";
 
-import type { DiagnosticReport, DiagnosticReportState } from "@/api/types";
+import type { DiagnosticReport, DiagnosticReportState, DiagnosticReportSummary } from "@/api/types";
 import { ConfirmDialog } from "@/components/ConfirmDialog";
 import { Badge } from "@/components/ui/badge";
 import { Button } from "@/components/ui/button";
@@ -417,7 +417,7 @@ function DiagnosticReportRow({
   report,
   onSelect,
 }: {
-  report: DiagnosticReport;
+  report: DiagnosticReportSummary;
   onSelect: (id: string) => void;
 }) {
   function handleKeyDown(event: KeyboardEvent<HTMLTableRowElement>) {
@@ -449,7 +449,7 @@ function DiagnosticReportRow({
         <div>{formatPlatform(report.platform)}</div>
         <div className="text-muted-foreground text-xs">
           v{report.app_version}
-          {report.manifest.report.app_build ? ` (${report.manifest.report.app_build})` : ""}
+          {report.app_build ? ` (${report.app_build})` : ""}
         </div>
       </TableCell>
       <TableCell>
@@ -484,7 +484,7 @@ function DiagnosticReportDetail({
   onDelete: () => void;
 }) {
   const device = report.manifest.device_summary;
-  const appBuild = report.manifest.report.app_build;
+  const appBuild = report.app_build;
 
   return (
     <div className="space-y-6">