diff --git a/.changeset/bucket-storage-report.md b/.changeset/bucket-storage-report.md
new file mode 100644
index 000000000..b37a614d3
--- /dev/null
+++ b/.changeset/bucket-storage-report.md
@@ -0,0 +1,8 @@
+---
+'@powersync/service-core': minor
+'@powersync/service-types': minor
+'@powersync/service-module-mongodb-storage': minor
+'@powersync/service-core-tests': minor
+---
+
+Add a `POST /api/admin/v1/bucket-report` admin endpoint reporting operations vs rows per bucket (MongoDB storage).
diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
index 5765f8b21..ef84b02db 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
@@ -22,7 +22,12 @@ import {
   utils,
   WatchWriteCheckpointOptions
 } from '@powersync/service-core';
-import { HydratedSyncConfig, ParameterLookupRows, ScopedParameterLookup } from '@powersync/service-sync-rules';
+import {
+  BucketDefinitionId,
+  HydratedSyncConfig,
+  ParameterLookupRows,
+  ScopedParameterLookup
+} from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 import { LRUCache } from 'lru-cache';
 import * as timers from 'timers/promises';
@@ -30,7 +35,7 @@ import { DEFAULT_CLEAR_BATCH_THROTTLE_RATE } from '../../types/types.js';
 import { MongoBucketStorage } from '../MongoBucketStorage.js';
 import { DEFAULT_INLINE_THRESHOLD_BYTES } from './common/PersistedBatch.js';
 import type { VersionedPowerSyncMongo } from './db.js';
-import { StorageConfig } from './models.js';
+import { BucketStateDocumentBase, StorageConfig } from './models.js';
 import { MongoBucketBatchOptions } from './MongoBucketBatch.js';
 import { MongoChecksumOptions, MongoChecksums } from './MongoChecksums.js';
 import { MongoCompactOptions, MongoCompactor } from './MongoCompactor.js';
@@ -66,6 +71,71 @@ interface InternalCheckpointChanges extends CheckpointChanges {
  */
 const CHECKPOINT_TIMEOUT_MS = 60_000;
 
+/**
+ * Above this many buckets, the report ranks a bounded `$sample` of bucket_state rather than every bucket, so
+ * the request cannot exhaust memory or run unbounded. Below it, the ranking is exact.
+ */
+const BUCKET_SELECTION_SAMPLE_THRESHOLD = 50_000;
+
+/** Number of buckets to sample when over {@link BUCKET_SELECTION_SAMPLE_THRESHOLD}. */
+const BUCKET_SELECTION_SAMPLE_SIZE = 10_000;
+
+/**
+ * Fewest operations sampled per bucket when estimating its row count. Buckets with fewer operations than
+ * this are read in full (exact).
+ */
+const BUCKET_ROW_SAMPLE_MIN = 1_000;
+
+/**
+ * Most operations sampled per bucket, capping the per-bucket cost on very large buckets at the price of a
+ * weaker estimate for buckets that are both extremely wide and barely fragmented (see {@link bucketRowSampleTarget}).
+ */
+const BUCKET_ROW_SAMPLE_MAX = 25_000;
+
+/** Maximum number of per-bucket row-estimate queries to run concurrently while building a report. */
+const BUCKET_ROW_SAMPLE_CONCURRENCY = 10;
+
+/** Maximum number of tables listed per bucket or definition in the report. */
+const BUCKET_REPORT_TABLE_LIMIT = 10;
+
+/** A worst-offender bucket selected from bucket_state, with the version-specific context needed to sample it. */
+export interface TopBucketCandidate {
+  bucket: string;
+  operations: number;
+  operationBytes: number;
+  /** v3 only: the bucket definition id, used to locate its per-definition bucket_data collection. */
+  defId?: BucketDefinitionId;
+}
+
+/** A bucket definition aggregated from bucket_state, with the context needed to sample its rows. */
+export interface TopDefinitionCandidate {
+  /** Definition name as it prefixes bucket names, e.g. `1#by_user`. */
+  definition: string;
+  bucketCount: number;
+  operations: number;
+  operationBytes: number;
+  /** v3 only: the bucket definition id, used to locate its per-definition bucket_data collection. */
+  defId?: BucketDefinitionId;
+}
+
+export interface TopBucketSelection {
+  buckets: TopBucketCandidate[];
+  definitions: TopDefinitionCandidate[];
+  /** True if more definitions exist than `definitions` holds ({@link storage.BUCKET_REPORT_DEFINITION_LIMIT}). */
+  definitionsTruncated: boolean;
+  totals: storage.BucketReportTotals;
+}
+
+export interface BucketRowEstimate {
+  rows: number;
+  /** Operations carrying a row identity (PUT/REMOVE), i.e. excluding MOVE/CLEAR compaction residue. */
+  rowOperations: number;
+  /** True if `rows` and `rowOperations` are sampled estimates rather than exact counts. */
+  estimated: boolean;
+  /** Tables in the (sampled) row-bearing history, ordered by their share of it, largest first. */
+  tables: string[];
+}
+
 export abstract class MongoSyncBucketStorage
   extends BaseObserver<storage.SyncRulesBucketStorageListener>
   implements storage.SyncRulesBucketStorage
@@ -377,6 +447,359 @@ export abstract class MongoSyncBucketStorage
     }
   }
 
+  async getBucketReport(options?: storage.GetBucketReportOptions): Promise<storage.BucketReport> {
+    const limit = storage.resolveBucketReportLimit(options?.limit);
+    try {
+      // Rank the worst-offender buckets, the per-definition rollup, and total operations from the
+      // pre-aggregated bucket state (bounded, in the database), then estimate each returned bucket's and
+      // definition's row count by sampling its operation history.
+      const { buckets, definitions, definitionsTruncated, totals } = await this.collectTopBuckets(limit);
+      // Each row estimate is an independent query; run a bounded number concurrently so the report cost
+      // scales with the limit without firing one query per bucket serially. Definitions sample their whole
+      // history and are the slowest jobs, so dispatch them first to overlap with the per-bucket estimates.
+      const rankedBuckets: storage.RankedBucketInput[] = new Array(buckets.length);
+      const rankedDefinitions: storage.RankedDefinitionInput[] = new Array(definitions.length);
+      const jobs = buckets.length + definitions.length;
+      let cursor = 0;
+      const runWorker = async () => {
+        while (true) {
+          const index = cursor++;
+          if (index >= jobs) {
+            return;
+          }
+          if (index < definitions.length) {
+            const candidate = definitions[index];
+            // A definition's row sample reads its whole (sampled) history, which on a very large instance
+            // can exceed the time budget even when the per-bucket estimates are fine. The rollup is
+            // supplementary: omit the definition rather than failing the whole report.
+            try {
+              const estimate = await this.estimateDefinitionRows(candidate);
+              rankedDefinitions[index] = {
+                definition: candidate.definition,
+                bucketCount: candidate.bucketCount,
+                operations: candidate.operations,
+                operationBytes: candidate.operationBytes,
+                rows: estimate.rows,
+                rowOperations: estimate.rowOperations,
+                rowsEstimated: estimate.estimated,
+                tables: estimate.tables
+              };
+            } catch (e) {
+              this.logger.warn(
+                `Skipping bucket report rollup for definition ${candidate.definition}: row sampling failed`,
+                e
+              );
+            }
+          } else {
+            const candidate = buckets[index - definitions.length];
+            const estimate = await this.estimateBucketRows(candidate);
+            rankedBuckets[index - definitions.length] = {
+              bucket: candidate.bucket,
+              operations: candidate.operations,
+              operationBytes: candidate.operationBytes,
+              rows: estimate.rows,
+              rowOperations: estimate.rowOperations,
+              rowsEstimated: estimate.estimated,
+              tables: estimate.tables
+            };
+          }
+        }
+      };
+      const workers = Math.min(BUCKET_ROW_SAMPLE_CONCURRENCY, jobs);
+      await Promise.all(Array.from({ length: workers }, () => runWorker()));
+      const sampledDefinitions = rankedDefinitions.filter((d) => d != null);
+      return storage.assembleBucketReport(
+        rankedBuckets,
+        sampledDefinitions,
+        totals,
+        // The rollup is also incomplete if a definition was dropped because sampling it failed.
+        definitionsTruncated || sampledDefinitions.length < definitions.length
+      );
+    } catch (e) {
+      // Translate a storage query timeout (maxTimeMS) into a specific, retryable error code rather than a
+      // generic internal error.
+      throw lib_mongo.mapQueryError(e, 'while building the bucket report');
+    }
+  }
+
+  /**
+   * Select the worst-offender buckets (by operation count), the per-definition rollup, and instance-wide
+   * operation totals from the pre-aggregated bucket state. Ranking and limiting happen in the database, so
+   * memory stays bounded. Implementations supply their version-specific bucket state collection and
+   * active-config filter.
+   */
+  protected abstract collectTopBuckets(limit: number): Promise<TopBucketSelection>;
+
+  /**
+   * Estimate a single bucket's live row count by sampling its operation history. Implementations differ
+   * because v1/v2 store one document per operation while v3 batches operations per document.
+   */
+  protected abstract estimateBucketRows(candidate: TopBucketCandidate): Promise<BucketRowEstimate>;
+
+  /**
+   * Estimate a whole definition's row count (a row counted once per bucket containing it) by sampling the
+   * definition's operation history, exactly like {@link estimateBucketRows} but at definition grain.
+   */
+  protected abstract estimateDefinitionRows(candidate: TopDefinitionCandidate): Promise<BucketRowEstimate>;
+
+  /**
+   * Rank buckets by operation count in the database and compute instance-wide operation totals, reading the
+   * pre-aggregated bucket state (compacted_state + estimate_since_compact). One document per bucket, no scan
+   * of bucket data.
+   *
+   * For very large bucket sets the candidates are drawn from a bounded `$sample` rather than the whole
+   * collection (so the request cannot run unbounded or exhaust memory), and the totals are scaled from the
+   * sample and flagged estimated. `allowDiskUse: false` makes an over-threshold exact attempt fail fast
+   * rather than spill to disk and degrade the live instance.
+   *
+   * Note: for v1/v2 storage, bucket_state is not backfilled (see models.ts: "only populated by new updates"),
+   * so buckets that predate bucket_state tracking and have not been updated or compacted since are missing
+   * here and under-counted. v3 always has bucket_state.
+   */
+  protected async aggregateTopBuckets<T extends BucketStateDocumentBase>(
+    collection: mongo.Collection<T>,
+    match: mongo.Filter<T>,
+    limit: number
+  ): Promise<{
+    buckets: { id: T['_id']; operations: number; operationBytes: number }[];
+    definitions: TopDefinitionCandidate[];
+    definitionsTruncated: boolean;
+    totals: storage.BucketReportTotals;
+  }> {
+    const operations = {
+      $add: [{ $ifNull: ['$compacted_state.count', 0] }, { $ifNull: ['$estimate_since_compact.count', 0] }]
+    };
+    const operationBytes = {
+      $add: [
+        { $toDouble: { $ifNull: ['$compacted_state.bytes', 0] } },
+        { $toDouble: { $ifNull: ['$estimate_since_compact.bytes', 0] } }
+      ]
+    };
+    // Bucket names are `<definition>[<serialized parameters>]`, so everything before the first `[` groups a
+    // bucket into its definition. v3 additionally carries the definition id in `_id.d`; `$first` is exact
+    // because all buckets sharing a name prefix share the definition (undefined for v1/v2).
+    const definitionKey = { $arrayElemAt: [{ $split: ['$_id.b', '['] }, 0] };
+
+    // estimatedDocumentCount is O(1) but ignores the match filter, so this is an upper bound on the active
+    // bucket count. That is fine for the sampling decision: over-estimating only switches to sampling sooner.
+    // It must NOT be used to scale the sampled totals though - the collection can hold buckets outside the
+    // match (other replication groups for v1/v2, inactive definitions for v3), which would over-scale.
+    const estimatedTotalBuckets = await collection.estimatedDocumentCount();
+    const sampled = estimatedTotalBuckets > BUCKET_SELECTION_SAMPLE_THRESHOLD;
+
+    const pipeline: mongo.Document[] = [{ $match: match }];
+    if (sampled) {
+      pipeline.push({ $sample: { size: BUCKET_SELECTION_SAMPLE_SIZE } });
+    }
+    pipeline.push({
+      $facet: {
+        totals: [
+          {
+            $group: {
+              _id: null,
+              operations: { $sum: operations },
+              operationBytes: { $sum: operationBytes },
+              bucketCount: { $sum: 1 }
+            }
+          }
+        ],
+        top: [{ $project: { _id: 1, operations, operationBytes } }, { $sort: { operations: -1 } }, { $limit: limit }],
+        definitions: [
+          {
+            $group: {
+              _id: definitionKey,
+              operations: { $sum: operations },
+              operationBytes: { $sum: operationBytes },
+              bucketCount: { $sum: 1 },
+              defId: { $first: '$_id.d' }
+            }
+          },
+          { $sort: { operations: -1 } },
+          // One past the cap: an extra result only signals that the rollup was truncated.
+          { $limit: storage.BUCKET_REPORT_DEFINITION_LIMIT + 1 }
+        ]
+      }
+    });
+
+    type FacetResult = {
+      totals: { operations: number; operationBytes: number; bucketCount: number }[];
+      top: { _id: T['_id']; operations: number; operationBytes: number }[];
+      definitions: {
+        _id: string;
+        operations: number;
+        operationBytes: number;
+        bucketCount: number;
+        defId?: BucketDefinitionId;
+      }[];
+    };
+    const [result] = await collection
+      .aggregate<FacetResult>(pipeline, { allowDiskUse: false, maxTimeMS: storage.BUCKET_REPORT_TIMEOUT_MS })
+      .toArray();
+
+    const rawTotals = result?.totals[0] ?? { operations: 0, operationBytes: 0, bucketCount: 0 };
+    const buckets = (result?.top ?? []).map((doc) => ({
+      id: doc._id,
+      operations: doc.operations,
+      operationBytes: doc.operationBytes
+    }));
+    const rawDefinitions = result?.definitions ?? [];
+    const definitionsTruncated = rawDefinitions.length > storage.BUCKET_REPORT_DEFINITION_LIMIT;
+    const mapDefinitions = (scale: number): TopDefinitionCandidate[] =>
+      rawDefinitions.slice(0, storage.BUCKET_REPORT_DEFINITION_LIMIT).map((d) => ({
+        definition: d._id,
+        bucketCount: Math.round(d.bucketCount * scale),
+        operations: Math.round(d.operations * scale),
+        operationBytes: Math.round(d.operationBytes * scale),
+        defId: d.defId
+      }));
+
+    if (!sampled) {
+      return {
+        buckets,
+        definitions: mapDefinitions(1),
+        definitionsTruncated,
+        totals: {
+          bucketCount: rawTotals.bucketCount,
+          operations: rawTotals.operations,
+          operationBytes: rawTotals.operationBytes,
+          estimated: false
+        }
+      };
+    }
+
+    // Scale the sampled totals up to the full *matched* set. countDocuments respects the match filter (so it
+    // excludes other groups / inactive definitions) and uses the _id index; it only runs on the already-large
+    // sampled path, and is bounded by maxTimeMS like the rest of the report. When the matched set fits within
+    // the sample, rawTotals is already exact and the scale collapses to 1. The sample is uniform across
+    // buckets, so the per-definition sums scale by the same factor; a definition small enough to be missed
+    // by the sample entirely is absent.
+    const matchedBuckets = await collection.countDocuments(match, { maxTimeMS: storage.BUCKET_REPORT_TIMEOUT_MS });
+    const scale = matchedBuckets / Math.max(rawTotals.bucketCount, 1);
+    return {
+      buckets,
+      definitions: mapDefinitions(scale),
+      definitionsTruncated,
+      totals: {
+        bucketCount: matchedBuckets,
+        operations: Math.round(rawTotals.operations * scale),
+        operationBytes: Math.round(rawTotals.operationBytes * scale),
+        estimated: true
+      }
+    };
+  }
+
+  /**
+   * Estimate a bucket's (or definition's) live rows from a sample of its operations.
+   *
+   * `buildPrefix(applySample)` returns a pipeline prefix that selects the operations (down-sampled when
+   * `applySample` is true) and yields documents with top-level `op`, `table` and `row_id` fields. Returns
+   * the distinct row count (exact when the whole history was read, otherwise estimated via
+   * {@link storage.estimateDistinctRows}); fragmentation is then `operations / rows`.
+   *
+   * `rowKey` is the `$group` key that identifies a row. Per-bucket estimates use the default (the bucket is
+   * fixed by the prefix); definition-level estimates must include the bucket name so a row is counted once
+   * per bucket containing it.
+   */
+  protected async estimateRowsFromOperationSample<T extends mongo.Document>(
+    collection: mongo.Collection<T>,
+    buildPrefix: (applySample: boolean) => mongo.Document[],
+    operations: number,
+    sampled: boolean,
+    rowKey: mongo.Document = { table: '$table', row_id: '$row_id' }
+  ): Promise<BucketRowEstimate> {
+    const runCounts = async (applySample: boolean) => {
+      const pipeline: mongo.Document[] = [
+        ...buildPrefix(applySample),
+        {
+          $facet: {
+            sampledOps: [{ $count: 'count' }],
+            rowOps: [{ $match: { op: { $in: ['PUT', 'REMOVE'] } } }, { $count: 'count' }],
+            distinctRows: [
+              { $match: { op: { $in: ['PUT', 'REMOVE'] } } },
+              { $group: { _id: rowKey } },
+              { $count: 'count' }
+            ],
+            tables: [
+              { $match: { op: { $in: ['PUT', 'REMOVE'] } } },
+              { $group: { _id: '$table', operations: { $sum: 1 } } },
+              { $sort: { operations: -1 } },
+              { $limit: BUCKET_REPORT_TABLE_LIMIT }
+            ]
+          }
+        }
+      ];
+      type FacetResult = {
+        sampledOps: { count: number }[];
+        rowOps: { count: number }[];
+        distinctRows: { count: number }[];
+        tables: { _id: string }[];
+      };
+      const [result] = await collection
+        .aggregate<FacetResult>(pipeline, { allowDiskUse: false, maxTimeMS: storage.BUCKET_REPORT_TIMEOUT_MS })
+        .toArray();
+      return {
+        sampledOps: result?.sampledOps[0]?.count ?? 0,
+        rowOps: result?.rowOps[0]?.count ?? 0,
+        distinctRows: result?.distinctRows[0]?.count ?? 0,
+        tables: (result?.tables ?? []).map((t) => t._id)
+      };
+    };
+
+    let counts = await runCounts(sampled);
+    if (sampled && counts.sampledOps == 0) {
+      // A document-level `$sampleRate` can select nothing when a bucket spans very few storage documents
+      // (v3 batches operations into a document). Fall back to an exact read so the bucket is not reported as
+      // zero rows. This reads the whole bucket only in the rare empty-sample case, which cannot happen for a
+      // bucket large enough to span many documents.
+      const exact = await runCounts(false);
+      return { rows: exact.distinctRows, rowOperations: exact.rowOps, estimated: false, tables: exact.tables };
+    }
+    if (counts.distinctRows == 0) {
+      // Nothing row-bearing was found (e.g. a bucket of only MOVE/CLEAR ops): treat as fully fragmented.
+      return { rows: 0, rowOperations: 0, estimated: sampled, tables: [] };
+    }
+    if (!sampled) {
+      // Read in full: the distinct row count is exact.
+      return { rows: counts.distinctRows, rowOperations: counts.rowOps, estimated: false, tables: counts.tables };
+    }
+    // Only PUT/REMOVE operations carry a row identity; MOVE/CLEAR (produced by compaction) do not. Run the
+    // estimator over the row-bearing operations only, scaling the bucket's operation count by the row-bearing
+    // share observed in the sample. Including identity-less operations in the model under-counts rows on
+    // compacted buckets. For uncompacted buckets rowOps equals sampledOps and this changes nothing.
+    const rowBearingOperations = Math.round(operations * (counts.rowOps / counts.sampledOps));
+    return {
+      rows: storage.estimateDistinctRows(rowBearingOperations, counts.rowOps, counts.distinctRows),
+      rowOperations: rowBearingOperations,
+      estimated: true,
+      tables: counts.tables
+    };
+  }
+
+  /**
+   * How many operations to sample when estimating a bucket's row count.
+   *
+   * {@link storage.estimateDistinctRows} infers the row count from how often the sample lands on the same
+   * row twice, so the sample must be large enough to contain such repeats. Sampling `sqrt(200 * operations)`
+   * operations yields on the order of 100 expected repeats even in the worst case of one row per operation,
+   * which keeps the estimate stable instead of swinging with sampling noise. The clamp bounds per-bucket
+   * cost; past the cap only very wide, barely fragmented buckets lose accuracy, and those are not the
+   * offenders the report exists to surface.
+   */
+  protected bucketRowSampleTarget(operations: number): number {
+    const target = Math.ceil(Math.sqrt(200 * operations));
+    return Math.min(BUCKET_ROW_SAMPLE_MAX, Math.max(BUCKET_ROW_SAMPLE_MIN, target));
+  }
+
+  /** Whether a bucket with this many operations should be sampled rather than read in full. */
+  protected shouldSampleBucketRows(operations: number): boolean {
+    return operations > this.bucketRowSampleTarget(operations);
+  }
+
+  /** `$sampleRate` for sampling roughly {@link bucketRowSampleTarget} operations from a bucket. */
+  protected bucketRowSampleRate(operations: number): number {
+    return this.bucketRowSampleTarget(operations) / operations;
+  }
+
   /**
    * The highest op id persisted for this stream, whether or not covered by a checkpoint.
    *
diff --git a/modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts b/modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts
index 89fec0c47..7470d42e3 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts
@@ -32,7 +32,14 @@ import { MongoChecksums } from '../MongoChecksums.js';
 import { MongoCompactOptions, MongoCompactor } from '../MongoCompactor.js';
 import { MongoParameterCompactor } from '../MongoParameterCompactor.js';
 import { MongoPersistedReplicationStream } from '../MongoPersistedReplicationStream.js';
-import { MongoSyncBucketStorage, MongoSyncBucketStorageOptions } from '../MongoSyncBucketStorage.js';
+import {
+  BucketRowEstimate,
+  MongoSyncBucketStorage,
+  MongoSyncBucketStorageOptions,
+  TopBucketCandidate,
+  TopBucketSelection,
+  TopDefinitionCandidate
+} from '../MongoSyncBucketStorage.js';
 import {
   BucketDataDocumentV1,
   BucketDataKeyV1,
@@ -190,6 +197,75 @@ export class MongoSyncBucketStorageV1 extends MongoSyncBucketStorage {
     return new MongoCompactorV1(this, this.db, options);
   }
 
+  // For storage v1/v2, bucket state and bucket data are shared collections scoped by group (replication stream).
+  protected async collectTopBuckets(limit: number): Promise<TopBucketSelection> {
+    const { buckets, definitions, definitionsTruncated, totals } = await this.aggregateTopBuckets(
+      this.db.bucketStateV1,
+      { '_id.g': this.replicationStreamId },
+      limit
+    );
+    return {
+      buckets: buckets.map((b) => ({ bucket: b.id.b, operations: b.operations, operationBytes: b.operationBytes })),
+      definitions,
+      definitionsTruncated,
+      totals
+    };
+  }
+
+  protected estimateBucketRows(candidate: TopBucketCandidate): Promise<BucketRowEstimate> {
+    // v1/v2 store one document per operation, so a bucket's ops are an id-prefix range that can be sampled directly.
+    const sampled = this.shouldSampleBucketRows(candidate.operations);
+    const buildPrefix = (applySample: boolean): mongo.Document[] => {
+      // Range-match on the whole `_id` (g, b, o) so the {_id} index is used; a dotted `{'_id.g','_id.b'}` match
+      // cannot use the compound-object index and would scan the whole collection per bucket.
+      const prefix: mongo.Document[] = [
+        {
+          $match: {
+            _id: idPrefixFilter<{ g: number; b: string; o: unknown }>(
+              { g: this.replicationStreamId, b: candidate.bucket },
+              ['o']
+            )
+          }
+        }
+      ];
+      if (applySample) {
+        prefix.push({ $match: { $sampleRate: this.bucketRowSampleRate(candidate.operations) } });
+      }
+      return prefix;
+    };
+    return this.estimateRowsFromOperationSample(this.db.bucketDataV1, buildPrefix, candidate.operations, sampled);
+  }
+
+  protected estimateDefinitionRows(candidate: TopDefinitionCandidate): Promise<BucketRowEstimate> {
+    const sampled = this.shouldSampleBucketRows(candidate.operations);
+    const buildPrefix = (applySample: boolean): mongo.Document[] => {
+      // All of a definition's bucket names start with `<definition>[`, so an `_id` range on that string
+      // prefix selects exactly the definition's operations via the index. `\\` (0x5C) is the character
+      // after `[` (0x5B), so [`<definition>[`, `<definition>\\`) cannot include any other definition:
+      // a longer definition name would have to differ at or before the `[`.
+      const prefix: mongo.Document[] = [
+        {
+          $match: {
+            _id: {
+              $gte: { g: this.replicationStreamId, b: `${candidate.definition}[`, o: new bson.MinKey() },
+              $lt: { g: this.replicationStreamId, b: `${candidate.definition}\\`, o: new bson.MinKey() }
+            }
+          }
+        }
+      ];
+      if (applySample) {
+        prefix.push({ $match: { $sampleRate: this.bucketRowSampleRate(candidate.operations) } });
+      }
+      return prefix;
+    };
+    // Include the bucket name in the row key: at definition grain a row counts once per bucket holding it.
+    return this.estimateRowsFromOperationSample(this.db.bucketDataV1, buildPrefix, candidate.operations, sampled, {
+      b: '$_id.b',
+      table: '$table',
+      row_id: '$row_id'
+    });
+  }
+
   protected createMongoParameterCompactor(
     checkpoint: InternalOpId,
     options: storage.CompactOptions
diff --git a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts
index a4b313003..80fa7a152 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts
@@ -15,7 +15,7 @@ import {
 import { JSONBig } from '@powersync/service-jsonbig';
 import { ParameterLookupRows, ScopedParameterLookup, SqliteJsonRow } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
-import { mapOpEntry, readSingleBatch, setSessionSnapshotTime } from '../../../utils/util.js';
+import { idPrefixFilter, mapOpEntry, readSingleBatch, setSessionSnapshotTime } from '../../../utils/util.js';
 import { MongoBucketStorage } from '../../MongoBucketStorage.js';
 import { BucketDataDoc } from '../common/BucketDataDoc.js';
 import { MongoSyncBucketStorageCheckpoint } from '../common/MongoSyncBucketStorageCheckpoint.js';
@@ -23,7 +23,14 @@ import { MongoChecksums } from '../MongoChecksums.js';
 import { MongoCompactOptions, MongoCompactor } from '../MongoCompactor.js';
 import { MongoParameterCompactor } from '../MongoParameterCompactor.js';
 import { MongoPersistedReplicationStream } from '../MongoPersistedReplicationStream.js';
-import { MongoSyncBucketStorage, MongoSyncBucketStorageOptions } from '../MongoSyncBucketStorage.js';
+import {
+  BucketRowEstimate,
+  MongoSyncBucketStorage,
+  MongoSyncBucketStorageOptions,
+  TopBucketCandidate,
+  TopBucketSelection,
+  TopDefinitionCandidate
+} from '../MongoSyncBucketStorage.js';
 import { loadBucketDataDocument } from './bucket-format.js';
 import {
   BucketDataDocumentV3,
@@ -209,6 +216,73 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
     return new MongoCompactorV3(this, this.db, options);
   }
 
+  // For storage v3, bucket state is a per-stream collection and bucket data is split into per-definition collections.
+  // A replication stream can host multiple sync configs (active + processing + stopped, until cleanup runs), all
+  // sharing these collections. Scope to the active config's definition ids so the report excludes stale buckets
+  // from old/stopped definitions. `this.storageIds` is derived from the active config only (see getActiveSyncConfig).
+  protected async collectTopBuckets(limit: number): Promise<TopBucketSelection> {
+    const { buckets, definitions, definitionsTruncated, totals } = await this.aggregateTopBuckets(
+      this.db.bucketState(this.replicationStreamId),
+      { '_id.d': { $in: this.storageIds.bucketDefinitionIds } },
+      limit
+    );
+    return {
+      buckets: buckets.map((b) => ({
+        bucket: b.id.b,
+        operations: b.operations,
+        operationBytes: b.operationBytes,
+        defId: b.id.d
+      })),
+      definitions,
+      definitionsTruncated,
+      totals
+    };
+  }
+
+  protected estimateBucketRows(candidate: TopBucketCandidate): Promise<BucketRowEstimate> {
+    // v3 batches operations into documents (one doc holds an `ops` array), in a per-definition collection.
+    // Sample whole batch documents, then unwind to operation level so the shared estimator sees one doc per op.
+    const sampled = this.shouldSampleBucketRows(candidate.operations);
+    const collection = this.db.bucketData(this.replicationStreamId, candidate.defId!);
+    const buildPrefix = (applySample: boolean): mongo.Document[] => {
+      // Range-match on the whole `_id` (b, o) so the {_id} index is used; a dotted `{'_id.b': ...}` match
+      // cannot use the compound-object index and would scan the whole collection per bucket.
+      const prefix: mongo.Document[] = [
+        { $match: { _id: idPrefixFilter<{ b: string; o: unknown }>({ b: candidate.bucket }, ['o']) } }
+      ];
+      if (applySample) {
+        prefix.push({ $match: { $sampleRate: this.bucketRowSampleRate(candidate.operations) } });
+      }
+      prefix.push({ $unwind: '$ops' }, { $replaceRoot: { newRoot: '$ops' } });
+      return prefix;
+    };
+    return this.estimateRowsFromOperationSample(collection, buildPrefix, candidate.operations, sampled);
+  }
+
+  protected estimateDefinitionRows(candidate: TopDefinitionCandidate): Promise<BucketRowEstimate> {
+    // A definition's operations are exactly its per-definition bucket_data collection, so no match stage is
+    // needed. Keep the bucket name alongside each unwound operation: at definition grain a row counts once
+    // per bucket holding it.
+    const sampled = this.shouldSampleBucketRows(candidate.operations);
+    const collection = this.db.bucketData(this.replicationStreamId, candidate.defId!);
+    const buildPrefix = (applySample: boolean): mongo.Document[] => {
+      const prefix: mongo.Document[] = [];
+      if (applySample) {
+        prefix.push({ $match: { $sampleRate: this.bucketRowSampleRate(candidate.operations) } });
+      }
+      prefix.push(
+        { $unwind: '$ops' },
+        { $project: { b: '$_id.b', op: '$ops.op', table: '$ops.table', row_id: '$ops.row_id' } }
+      );
+      return prefix;
+    };
+    return this.estimateRowsFromOperationSample(collection, buildPrefix, candidate.operations, sampled, {
+      b: '$b',
+      table: '$table',
+      row_id: '$row_id'
+    });
+  }
+
   protected createMongoParameterCompactor(
     checkpoint: InternalOpId,
     options: storage.CompactOptions
diff --git a/modules/module-mongodb-storage/test/src/bucket-report-scoping.test.ts b/modules/module-mongodb-storage/test/src/bucket-report-scoping.test.ts
new file mode 100644
index 000000000..040e90cd9
--- /dev/null
+++ b/modules/module-mongodb-storage/test/src/bucket-report-scoping.test.ts
@@ -0,0 +1,128 @@
+import { MongoSyncBucketStorageV3 } from '@module/storage/implementation/v3/MongoSyncBucketStorageV3.js';
+import { storage, updateSyncRulesFromYaml } from '@powersync/service-core';
+import { test_utils } from '@powersync/service-core-tests';
+import * as bson from 'bson';
+import { describe, expect, test } from 'vitest';
+import { INITIALIZED_MONGO_STORAGE_FACTORY } from './util.js';
+
+function sourceDescriptor(name: string, objectId: string): storage.SourceEntityDescriptor {
+  return {
+    connectionTag: storage.SourceTable.DEFAULT_TAG,
+    objectId,
+    schema: 'public',
+    name,
+    replicaIdColumns: [{ name: 'id', type: 'VARCHAR', typeId: 25 }]
+  };
+}
+
+function objectIdGenerator(id: string) {
+  let used = false;
+  return () => {
+    if (used) {
+      throw new Error(`Can only generate a single id using ${id}`);
+    }
+    used = true;
+    return new bson.ObjectId(id);
+  };
+}
+
+/**
+ * In V3 a replication stream can host multiple sync configs (active + stopped, until cleanup runs), all sharing
+ * the per-stream bucket_state and source_records collections. The report must only include the active config's
+ * bucket definitions, not stale ones from a previous (now stopped) config.
+ */
+describe('bucket report scoping - mongodb v3', () => {
+  test('excludes buckets from stopped/old sync configs sharing the replication stream', async () => {
+    await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+
+    // Config 1: replicate todos. Writing a row creates a data bucket for this config.
+    const first = await factory.updateSyncRules(
+      updateSyncRulesFromYaml(
+        `
+config:
+  edition: 3
+
+streams:
+  by_owner:
+    query: SELECT * FROM todos WHERE owner_id = subscription.parameter('owner_id')
+`,
+        { storageVersion: 3 }
+      )
+    );
+    const firstStorage = factory.getInstance(first) as MongoSyncBucketStorageV3;
+    await using firstWriter = await firstStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const todosTable = (
+      await firstWriter.resolveTables({
+        connection_id: 1,
+        source: sourceDescriptor('todos', 'todos-relation'),
+        idGenerator: objectIdGenerator('6544e3899293153fa7b38360')
+      })
+    ).tables[0];
+    await firstWriter.save({
+      sourceTable: todosTable,
+      tag: storage.SaveOperationTag.INSERT,
+      after: { id: 'todo-1', owner_id: 'user-1' },
+      afterReplicaId: test_utils.rid('todo-1')
+    });
+    await firstWriter.markAllSnapshotDone('1/1');
+    await firstWriter.commit('1/1');
+    await firstWriter.flush();
+
+    // While config 1 is active, its bucket(s) show up in the report.
+    const firstReport = await firstStorage.getBucketReport();
+    expect(firstReport.totals.bucketCount).toBeGreaterThan(0);
+
+    // Config 2: a different stream over a different table. Config 1 transitions to STOP, but its bucket_state
+    // and source_records rows remain in the shared collections until cleanup runs (which we deliberately skip).
+    const second = await factory.updateSyncRules(
+      updateSyncRulesFromYaml(
+        `
+config:
+  edition: 3
+
+streams:
+  by_project:
+    query: SELECT * FROM scenes WHERE project_id = subscription.parameter('project_id')
+`,
+        { storageVersion: 3 }
+      )
+    );
+    expect(second.replicationStreamId).toBe(first.replicationStreamId);
+
+    // Drive config 2 to snapshot-done so it becomes ACTIVE and config 1 transitions to STOP (config 1 keeps
+    // serving until the new config finishes processing). Config 1's stale rows remain until cleanup, which we skip.
+    const replicatingStreams = await factory.getReplicatingReplicationStreams();
+    expect(replicatingStreams).toHaveLength(1);
+    const secondStorage = factory.getInstance(replicatingStreams[0]) as MongoSyncBucketStorageV3;
+    await using secondWriter = await secondStorage.createWriter(test_utils.BATCH_OPTIONS);
+    // Give config 2 its own replicated row, so the report has an active-config bucket to include.
+    const scenesTable = (
+      await secondWriter.resolveTables({
+        connection_id: 1,
+        source: sourceDescriptor('scenes', 'scenes-relation'),
+        idGenerator: objectIdGenerator('6544e3899293153fa7b38361')
+      })
+    ).tables[0];
+    await secondWriter.save({
+      sourceTable: scenesTable,
+      tag: storage.SaveOperationTag.INSERT,
+      after: { id: 'scene-1', project_id: 'project-1' },
+      afterReplicaId: test_utils.rid('scene-1')
+    });
+    await secondWriter.markAllSnapshotDone('2/1');
+    await secondWriter.commit('2/1');
+    await secondWriter.flush();
+
+    const activeConfig = await factory.getActiveSyncConfig();
+    expect(activeConfig).not.toBeNull();
+    const activeStorage = activeConfig!.storage as MongoSyncBucketStorageV3;
+    const secondReport = await activeStorage.getBucketReport();
+
+    // The active config's own bucket is included (include-active), while config 1's stale buckets, which still
+    // exist in the shared collections, are excluded (exclude-stale). Without scoping to the active config's
+    // definition ids, config 1's bucket would leak in here.
+    expect(secondReport.totals.bucketCount).toBeGreaterThan(0);
+    const firstBucketNames = new Set(firstReport.buckets.map((b) => b.bucket));
+    expect(secondReport.buckets.some((b) => firstBucketNames.has(b.bucket))).toBe(false);
+  });
+});
diff --git a/modules/module-mongodb-storage/test/src/storage.test.ts b/modules/module-mongodb-storage/test/src/storage.test.ts
index 5606af296..035399faf 100644
--- a/modules/module-mongodb-storage/test/src/storage.test.ts
+++ b/modules/module-mongodb-storage/test/src/storage.test.ts
@@ -20,6 +20,8 @@ for (let storageVersion of TEST_STORAGE_VERSIONS) {
   describe(`Mongo Sync Bucket Storage - Checkpoints - v${storageVersion}`, () =>
     register.registerDataStorageCheckpointTests({ ...INITIALIZED_MONGO_STORAGE_FACTORY, storageVersion }));
 
+  describe(`Mongo Sync Bucket Storage - Bucket report - v${storageVersion}`, () =>
+    register.registerBucketReportTests({ ...INITIALIZED_MONGO_STORAGE_FACTORY, storageVersion }));
   describe(`Mongo Sync Bucket Storage - write checkpoint metadata - v${storageVersion}`, () => {
     test('uses checkpoint_requested_at as the client-requested checkpoint marker', async () => {
       await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
diff --git a/packages/service-core-tests/src/tests/register-bucket-report-tests.ts b/packages/service-core-tests/src/tests/register-bucket-report-tests.ts
new file mode 100644
index 000000000..3c3d8e58b
--- /dev/null
+++ b/packages/service-core-tests/src/tests/register-bucket-report-tests.ts
@@ -0,0 +1,347 @@
+import { storage, updateSyncRulesFromYaml } from '@powersync/service-core';
+import { expect, test } from 'vitest';
+import * as test_utils from '../test-utils/test-utils-index.js';
+
+/**
+ * Tests for {@link storage.SyncRulesBucketStorage.getBucketReport}: per-bucket operations vs live rows.
+ *
+ * Asserts on stable counts (operations, rows, fragmentation, operation totals) rather than op_ids or
+ * checksums, which differ between storage backends and versions. The buckets here are tiny (well under the
+ * row-sample target), so row counts are exact (`rowsEstimated: false`); the sampling path is exercised in
+ * the higher-volume manual tests.
+ */
+export function registerBucketReportTests(config: storage.TestStorageConfig) {
+  const generateStorageFactory = config.factory;
+  const storageVersion = config.storageVersion ?? storage.CURRENT_STORAGE_VERSION;
+
+  const GLOBAL_SYNC_RULES = `
+bucket_definitions:
+  global:
+    data: [select * from test]
+`;
+
+  // A constant parameter query keeps op_ids stable across backends (no bucket_parameter records); the data
+  // query routes each row into a bucket keyed by its own `b` value, so rows land in grouped["b1"]/grouped["b2"].
+  const GROUPED_SYNC_RULES = ` bucket_definitions:
+    grouped:
+      parameters: select 'b' as b
+      data:
+        - select * from test where b = bucket.b`;
+
+  const getReport = (bucketStorage: storage.SyncRulesBucketStorage, options?: storage.GetBucketReportOptions) => {
+    if (bucketStorage.getBucketReport == null) {
+      throw new Error('Storage backend does not implement getBucketReport');
+    }
+    return bucketStorage.getBucketReport(options);
+  };
+
+  test('reports operations and live rows for a single bucket', async () => {
+    await using factory = await generateStorageFactory();
+    const { stream, content } = await test_utils.deploySyncRules(
+      factory,
+      updateSyncRulesFromYaml(GLOBAL_SYNC_RULES, { storageVersion })
+    );
+    const bucketStorage = factory.getInstance(stream);
+
+    await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config);
+    await writer.markAllSnapshotDone('1/1');
+    for (const id of ['t1', 't2', 't3']) {
+      await writer.save({
+        sourceTable: testTable,
+        tag: storage.SaveOperationTag.INSERT,
+        after: { id },
+        afterReplicaId: test_utils.rid(id)
+      });
+    }
+    await writer.commit('1/1');
+    await writer.flush();
+
+    const bucket = test_utils.bucketRequest(content, 'global[]').bucket;
+    const report = await getReport(bucketStorage);
+
+    expect(report.totals.bucketCount).toEqual(1);
+    expect(report.bucketsTruncated).toEqual(false);
+    expect(report.definitionsTruncated).toEqual(false);
+
+    const stats = report.buckets.find((b) => b.bucket === bucket)!;
+    // Three inserts of distinct ids: three operations, three live rows, fully compacted (ratio 1).
+    expect(stats).toMatchObject({
+      operations: 3,
+      rows: 3,
+      fragmentation: 1,
+      rowsEstimated: false,
+      suggestedAction: 'none',
+      tables: ['test']
+    });
+    expect(stats.operationBytes).toBeGreaterThan(0);
+    expect(report.totals).toMatchObject({ operations: 3, estimated: false });
+
+    // The definition rollup aggregates the single bucket. The definition name is the bucket-name prefix.
+    expect(report.definitions).toHaveLength(1);
+    expect(report.definitions[0]).toMatchObject({
+      definition: bucket.split('[')[0],
+      bucketCount: 1,
+      operations: 3,
+      rows: 3,
+      fragmentation: 1,
+      suggestedAction: 'none',
+      tables: ['test']
+    });
+  });
+
+  test('operations exceed live rows after updates, and compaction reduces fragmentation', async () => {
+    await using factory = await generateStorageFactory();
+    const { stream, content } = await test_utils.deploySyncRules(
+      factory,
+      updateSyncRulesFromYaml(GLOBAL_SYNC_RULES, { storageVersion })
+    );
+    const bucketStorage = factory.getInstance(stream);
+
+    await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config);
+    await writer.markAllSnapshotDone('1/1');
+    // Two rows, each inserted then updated twice: six operations over two live rows.
+    for (const id of ['t1', 't2']) {
+      for (const value of ['a', 'b', 'c']) {
+        await writer.save({
+          sourceTable: testTable,
+          tag: value === 'a' ? storage.SaveOperationTag.INSERT : storage.SaveOperationTag.UPDATE,
+          after: { id, value },
+          afterReplicaId: test_utils.rid(id)
+        });
+      }
+    }
+    await writer.commit('1/1');
+    await writer.flush();
+
+    const bucket = test_utils.bucketRequest(content, 'global[]').bucket;
+
+    const before = await getReport(bucketStorage);
+    const beforeStats = before.buckets.find((b) => b.bucket === bucket)!;
+    expect(beforeStats).toMatchObject({ operations: 6, rows: 2, fragmentation: 3 });
+
+    await bucketStorage.compact({
+      clearBatchLimit: 10,
+      moveBatchLimit: 10,
+      moveBatchQueryLimit: 10,
+      minBucketChanges: 1,
+      minChangeRatio: 0
+    });
+
+    const after = await getReport(bucketStorage);
+    const afterStats = after.buckets.find((b) => b.bucket === bucket)!;
+    // Live rows are unchanged; the operation history shrinks toward the live row count.
+    expect(afterStats.rows).toEqual(2);
+    expect(afterStats.operations).toBeLessThan(beforeStats.operations);
+    expect(afterStats.operations).toBeGreaterThanOrEqual(afterStats.rows);
+    expect(afterStats.fragmentation).toBeLessThan(beforeStats.fragmentation);
+  });
+
+  test('reports every bucket, ranks worst-first, and totals across all buckets', async () => {
+    await using factory = await generateStorageFactory();
+    const { stream, content } = await test_utils.deploySyncRules(
+      factory,
+      updateSyncRulesFromYaml(GROUPED_SYNC_RULES, { storageVersion })
+    );
+    const bucketStorage = factory.getInstance(stream);
+
+    await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config);
+    await writer.markAllSnapshotDone('1/1');
+    // grouped["b1"]: one row, three operations (insert + two updates).
+    for (const value of ['a', 'b', 'c']) {
+      await writer.save({
+        sourceTable: testTable,
+        tag: value === 'a' ? storage.SaveOperationTag.INSERT : storage.SaveOperationTag.UPDATE,
+        after: { id: 't1', b: 'b1', value },
+        afterReplicaId: test_utils.rid('t1')
+      });
+    }
+    // grouped["b2"]: two rows, two operations.
+    for (const id of ['t2', 't3']) {
+      await writer.save({
+        sourceTable: testTable,
+        tag: storage.SaveOperationTag.INSERT,
+        after: { id, b: 'b2' },
+        afterReplicaId: test_utils.rid(id)
+      });
+    }
+    await writer.commit('1/1');
+    await writer.flush();
+
+    const b1 = test_utils.bucketRequest(content, 'grouped["b1"]').bucket;
+    const b2 = test_utils.bucketRequest(content, 'grouped["b2"]').bucket;
+
+    const report = await getReport(bucketStorage);
+    expect(report.totals.bucketCount).toEqual(2);
+    expect(report.totals).toMatchObject({ operations: 5, estimated: false });
+
+    // Ranked worst-first by operation count: b1 (3) before b2 (2).
+    expect(report.buckets.map((b) => b.bucket)).toEqual([b1, b2]);
+    expect(report.buckets.find((b) => b.bucket === b1)).toMatchObject({ operations: 3, rows: 1 });
+    expect(report.buckets.find((b) => b.bucket === b2)).toMatchObject({ operations: 2, rows: 2 });
+
+    // Both buckets belong to one definition; the rollup sums them, counting each bucket's rows separately.
+    expect(report.definitions).toHaveLength(1);
+    expect(report.definitions[0]).toMatchObject({
+      definition: b1.split('[')[0],
+      bucketCount: 2,
+      operations: 5,
+      rows: 3
+    });
+
+    // operationBytes is an aggregated ($toDouble) sum; assert every bucket is non-zero and that the
+    // per-bucket bytes add up to the instance total.
+    expect(report.totals.operationBytes).toBeGreaterThan(0);
+    for (const bucket of report.buckets) {
+      expect(bucket.operationBytes).toBeGreaterThan(0);
+    }
+    const summedBytes = report.buckets.reduce((total, bucket) => total + bucket.operationBytes, 0);
+    expect(summedBytes).toEqual(report.totals.operationBytes);
+  });
+
+  test('limit truncates the bucket list but totals still span all buckets', async () => {
+    await using factory = await generateStorageFactory();
+    const { stream, content } = await test_utils.deploySyncRules(
+      factory,
+      updateSyncRulesFromYaml(GROUPED_SYNC_RULES, { storageVersion })
+    );
+    const bucketStorage = factory.getInstance(stream);
+
+    await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config);
+    await writer.markAllSnapshotDone('1/1');
+    // grouped["b1"]: two operations; grouped["b2"]: one operation.
+    for (const value of ['a', 'b']) {
+      await writer.save({
+        sourceTable: testTable,
+        tag: value === 'a' ? storage.SaveOperationTag.INSERT : storage.SaveOperationTag.UPDATE,
+        after: { id: 't1', b: 'b1', value },
+        afterReplicaId: test_utils.rid('t1')
+      });
+    }
+    await writer.save({
+      sourceTable: testTable,
+      tag: storage.SaveOperationTag.INSERT,
+      after: { id: 't2', b: 'b2' },
+      afterReplicaId: test_utils.rid('t2')
+    });
+    await writer.commit('1/1');
+    await writer.flush();
+
+    const b1 = test_utils.bucketRequest(content, 'grouped["b1"]').bucket;
+
+    const report = await getReport(bucketStorage, { limit: 1 });
+    expect(report.bucketsTruncated).toEqual(true);
+    expect(report.buckets.map((b) => b.bucket)).toEqual([b1]);
+    // Totals still cover every bucket, not just the truncated list.
+    expect(report.totals.bucketCount).toEqual(2);
+    expect(report.totals).toMatchObject({ operations: 3, estimated: false });
+  });
+
+  test('caps the definition rollup and flags the truncation', async () => {
+    // Two definitions past the rollup cap; a single row lands in every definition's global bucket.
+    const definitionCount = storage.BUCKET_REPORT_DEFINITION_LIMIT + 2;
+    const manyDefinitions =
+      'bucket_definitions:\n' +
+      Array.from({ length: definitionCount }, (_, i) => `  def${i}:\n    data: [select * from test]\n`).join('');
+
+    await using factory = await generateStorageFactory();
+    const { stream } = await test_utils.deploySyncRules(
+      factory,
+      updateSyncRulesFromYaml(manyDefinitions, { storageVersion })
+    );
+    const bucketStorage = factory.getInstance(stream);
+
+    await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config);
+    await writer.markAllSnapshotDone('1/1');
+    await writer.save({
+      sourceTable: testTable,
+      tag: storage.SaveOperationTag.INSERT,
+      after: { id: 't1' },
+      afterReplicaId: test_utils.rid('t1')
+    });
+    await writer.commit('1/1');
+    await writer.flush();
+
+    const report = await getReport(bucketStorage);
+    expect(report.totals.bucketCount).toEqual(definitionCount);
+    expect(report.bucketsTruncated).toEqual(false);
+    expect(report.definitions).toHaveLength(storage.BUCKET_REPORT_DEFINITION_LIMIT);
+    expect(report.definitionsTruncated).toEqual(true);
+  });
+
+  test('samples the row count for a bucket above the sampling threshold', async () => {
+    await using factory = await generateStorageFactory();
+    const { stream, content } = await test_utils.deploySyncRules(
+      factory,
+      updateSyncRulesFromYaml(GLOBAL_SYNC_RULES, { storageVersion })
+    );
+    const bucketStorage = factory.getInstance(stream);
+
+    await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], config);
+    await writer.markAllSnapshotDone('1/1');
+
+    // 50 rows, each updated 25 times, is 1,300 operations against 50 live rows. That is past the 1,000
+    // operation threshold, so the report samples the operation history rather than reading it in full and
+    // the row count comes back as an estimate. The value per update varies so no two writes are identical.
+    // Each round is flushed separately so the operations span many storage documents, as they would in real
+    // replication (some backends batch operations per document, and a sample must see more than one).
+    const rowCount = 50;
+    const updatesPerRow = 25;
+    for (let row = 0; row < rowCount; row++) {
+      await writer.save({
+        sourceTable: testTable,
+        tag: storage.SaveOperationTag.INSERT,
+        after: { id: `r${row}` },
+        afterReplicaId: test_utils.rid(`r${row}`)
+      });
+    }
+    await writer.commit('1/1');
+    await writer.flush();
+    for (let update = 0; update < updatesPerRow; update++) {
+      for (let row = 0; row < rowCount; row++) {
+        await writer.save({
+          sourceTable: testTable,
+          tag: storage.SaveOperationTag.UPDATE,
+          after: { id: `r${row}`, value: `v${update}` },
+          afterReplicaId: test_utils.rid(`r${row}`)
+        });
+      }
+      await writer.commit('1/1');
+      await writer.flush();
+    }
+
+    const bucket = test_utils.bucketRequest(content, 'global[]').bucket;
+    const report = await getReport(bucketStorage);
+    const stats = report.buckets.find((b) => b.bucket === bucket)!;
+
+    // The operation count is exact (read from bucket_state); the row count is a sampled estimate.
+    expect(stats.operations).toEqual(rowCount + rowCount * updatesPerRow);
+    expect(stats.rowsEstimated).toEqual(true);
+    // The sample covers enough of a bucket this fragmented to recover the 50 live rows within a small margin.
+    expect(stats.rows).toBeGreaterThanOrEqual(45);
+    expect(stats.rows).toBeLessThanOrEqual(55);
+    // Fragmentation is operations / rows, so a heavily updated bucket reads well above 1.
+    expect(stats.fragmentation).toBeGreaterThan(10);
+    // The history is un-compacted superseded churn, which a compact reclaims.
+    expect(stats.suggestedAction).toEqual('compact');
+    // The sampled history names the tables a defragment would touch.
+    expect(stats.tables).toEqual(['test']);
+
+    // The definition rollup samples the same history at definition grain.
+    expect(report.definitions).toHaveLength(1);
+    const defStats = report.definitions[0];
+    expect(defStats).toMatchObject({
+      bucketCount: 1,
+      operations: stats.operations,
+      suggestedAction: 'compact',
+      tables: ['test']
+    });
+    expect(defStats.rows).toBeGreaterThanOrEqual(45);
+    expect(defStats.rows).toBeLessThanOrEqual(55);
+  });
+}
diff --git a/packages/service-core-tests/src/tests/tests-index.ts b/packages/service-core-tests/src/tests/tests-index.ts
index a40468a32..5be5e48f1 100644
--- a/packages/service-core-tests/src/tests/tests-index.ts
+++ b/packages/service-core-tests/src/tests/tests-index.ts
@@ -1,3 +1,4 @@
+export * from './register-bucket-report-tests.js';
 export * from './register-bucket-validation-tests.js';
 export * from './register-compacting-tests.js';
 export * from './register-data-storage-checkpoint-tests.js';
diff --git a/packages/service-core/src/routes/endpoints/admin.ts b/packages/service-core/src/routes/endpoints/admin.ts
index 5ba6ced06..f52aa8bc3 100644
--- a/packages/service-core/src/routes/endpoints/admin.ts
+++ b/packages/service-core/src/routes/endpoints/admin.ts
@@ -268,4 +268,76 @@ export const validate = routeDefinition({
   }
 });
 
-export const ADMIN_ROUTES = [executeSql, diagnostics, getSchema, reprocess, validate];
+/**
+ * Per-bucket report of total operations vs total live rows in storage, for the active sync config.
+ *
+ * Answers the recurring "why is my Data Synced so high" question. A high `operations / rows` ratio
+ * indicates fragmented buckets that a compact or defragment can reclaim.
+ */
+export const bucketReport = routeDefinition({
+  path: '/api/admin/v1/bucket-report',
+  method: router.HTTPMethod.POST,
+  authorize: authApi,
+  validator: schema.createTsCodecValidator(internal_routes.BucketReportRequest, { allowAdditional: true }),
+  handler: async (payload) => {
+    const {
+      context: { service_context }
+    } = payload;
+    const {
+      storageEngine: { activeBucketStorage }
+    } = service_context;
+
+    const active = await activeBucketStorage.getActiveSyncConfig();
+    if (active == null) {
+      throw new errors.ServiceError({
+        status: 422,
+        code: ErrorCode.PSYNC_S4104,
+        description: 'No active sync config'
+      });
+    }
+
+    if (active.storage.getBucketReport == null) {
+      throw new errors.ServiceError({
+        status: 422,
+        code: ErrorCode.PSYNC_S2001,
+        description: 'The configured storage provider does not support bucket reporting'
+      });
+    }
+
+    const report = await active.storage.getBucketReport({ limit: payload.params.limit });
+
+    return internal_routes.BucketReportResponse.encode({
+      buckets: report.buckets.map((bucket) => ({
+        bucket: bucket.bucket,
+        operations: bucket.operations,
+        rows: bucket.rows,
+        operation_bytes: bucket.operationBytes,
+        fragmentation: bucket.fragmentation,
+        rows_estimated: bucket.rowsEstimated,
+        suggested_action: bucket.suggestedAction,
+        tables: bucket.tables
+      })),
+      definitions: report.definitions.map((definition) => ({
+        definition: definition.definition,
+        bucket_count: definition.bucketCount,
+        operations: definition.operations,
+        operation_bytes: definition.operationBytes,
+        rows: definition.rows,
+        fragmentation: definition.fragmentation,
+        rows_estimated: definition.rowsEstimated,
+        suggested_action: definition.suggestedAction,
+        tables: definition.tables
+      })),
+      totals: {
+        bucket_count: report.totals.bucketCount,
+        operations: report.totals.operations,
+        operation_bytes: report.totals.operationBytes,
+        estimated: report.totals.estimated
+      },
+      buckets_truncated: report.bucketsTruncated,
+      definitions_truncated: report.definitionsTruncated
+    });
+  }
+});
+
+export const ADMIN_ROUTES = [executeSql, diagnostics, getSchema, reprocess, validate, bucketReport];
diff --git a/packages/service-core/src/storage/SyncRulesBucketStorage.ts b/packages/service-core/src/storage/SyncRulesBucketStorage.ts
index c626ccaef..a6454dea3 100644
--- a/packages/service-core/src/storage/SyncRulesBucketStorage.ts
+++ b/packages/service-core/src/storage/SyncRulesBucketStorage.ts
@@ -8,6 +8,7 @@ import {
 import * as bson from 'bson';
 import { PerformanceTracer } from '../tracing/PerformanceTracer.js';
 import * as util from '../util/util-index.js';
+import { BucketReport, GetBucketReportOptions } from './bucket-report.js';
 import { BucketStorageBatch, FlushedResult, SaveUpdate } from './BucketStorageBatch.js';
 import { BucketStorageFactory } from './BucketStorageFactory.js';
 import { ParsedSyncConfigSet } from './ParsedSyncConfigSet.js';
@@ -167,6 +168,17 @@ export interface SyncRulesBucketStorage
    * inside this replication stream.
    */
   cleanupStoppedSyncConfigs?(options: CleanupStoppedSyncConfigsOptions): Promise<CleanupStoppedSyncConfigsResult>;
+
+  /**
+   * Per-bucket report of total operations vs total live rows in storage.
+   *
+   * Intended for an on-demand admin/diagnostics view (e.g. answering "why is my Data Synced so high"),
+   * not as a live gauge. How the counts are derived is backend-specific, and the report may be relatively
+   * expensive on large instances.
+   *
+   * Optional: storage providers that don't implement it are reported as unsupported by the API.
+   */
+  getBucketReport?(options?: GetBucketReportOptions): Promise<BucketReport>;
 }
 
 export interface SyncRulesBucketStorageListener {
diff --git a/packages/service-core/src/storage/bucket-report.ts b/packages/service-core/src/storage/bucket-report.ts
new file mode 100644
index 000000000..d5920c20a
--- /dev/null
+++ b/packages/service-core/src/storage/bucket-report.ts
@@ -0,0 +1,312 @@
+/**
+ * Per-bucket storage report for an active sync config.
+ *
+ * - An **operation** is any entry in a bucket's append-only history (`PUT`, `REMOVE`, `MOVE`, `CLEAR`).
+ * - A **row** is a distinct live object currently in the bucket.
+ *
+ * A new client downloads every operation, not just live rows, so `operations / rows` is effectively a
+ * fragmentation / compaction-efficiency score: a fully compacted bucket trends towards ~1, while a high
+ * ratio is the usual cause of an unexpectedly high "Data Synced" metric and is reclaimable via compact/defragment.
+ *
+ * Scaling note: the report does NOT scan all storage. It ranks buckets by their pre-aggregated operation
+ * count and returns the worst offenders (top-N). Row counts (and therefore fragmentation) for those buckets
+ * are derived by sampling the operation history, so on large buckets they are estimates, flagged per bucket.
+ */
+
+/**
+ * Time budget for the per-bucket report's bucket-selection aggregation (`maxTimeMS`). Bounded so an admin
+ * request on a large instance fails fast instead of running unbounded.
+ */
+export const BUCKET_REPORT_TIMEOUT_MS: number = 60_000;
+
+/**
+ * Number of worst-offender buckets returned when the request omits a `limit`. Row counts are sampled per
+ * returned bucket, so this also bounds how much sampling work the report does.
+ */
+export const DEFAULT_BUCKET_REPORT_LIMIT: number = 50;
+
+/**
+ * Maximum number of bucket definitions in the report's definition rollup. Rows are sampled per returned
+ * definition (like per-bucket rows), so this bounds that sampling work. Configs rarely approach this many
+ * definitions.
+ */
+export const BUCKET_REPORT_DEFINITION_LIMIT: number = 20;
+
+/** Fragmentation below this is considered healthy: no maintenance action is suggested. */
+export const BUCKET_ACTION_MIN_FRAGMENTATION: number = 2;
+
+/**
+ * When at least this share of a bucket's operations is compaction residue (MOVE/CLEAR, no row identity),
+ * compaction has already done its work and only a defragment reduces what new clients download.
+ */
+export const BUCKET_ACTION_RESIDUE_SHARE: number = 0.5;
+
+/**
+ * When at least this share of a bucket's row-bearing operations (PUT/REMOVE) is superseded history (more
+ * operations than rows), a compact reclaims it (as MOVE conversions and a CLEAR prefix).
+ */
+export const BUCKET_ACTION_SUPERSEDED_SHARE: number = 0.5;
+
+/** Suggested maintenance action for a bucket or definition. See {@link suggestBucketAction}. */
+export type BucketAction = 'none' | 'compact' | 'defragment' | 'both';
+
+export interface BucketStorageStats {
+  /** Full bucket name, e.g. `by_user["u1"]`. */
+  bucket: string;
+  /** Total operations in the bucket's history. */
+  operations: number;
+  /** Live rows in the bucket. Exact for small buckets, otherwise a sampled estimate (see `rowsEstimated`). */
+  rows: number;
+  /** Approximate size of the operation history in bytes. */
+  operationBytes: number;
+  /**
+   * `operations / max(rows, 1)`. ~1 is healthy (fully compacted); higher means more operation-history
+   * overhead that a compact/defragment can reclaim.
+   */
+  fragmentation: number;
+  /** True if `rows` (and therefore `fragmentation`) is a sampled estimate rather than an exact count. */
+  rowsEstimated: boolean;
+  /** Suggested maintenance action derived from the operation mix. See {@link suggestBucketAction}. */
+  suggestedAction: BucketAction;
+  /**
+   * Tables making up the (sampled) operation history, ordered by their share of it, largest first. These
+   * are the tables whose rows a defragment should touch.
+   */
+  tables: string[];
+}
+
+/** Aggregated stats for one bucket definition (one `bucket_definitions` entry in the sync config). */
+export interface BucketDefinitionStats {
+  /** Definition name as it prefixes bucket names, e.g. `1#by_user` (versioned in storage v2 and later). */
+  definition: string;
+  /** Number of buckets in this definition with stored operations. */
+  bucketCount: number;
+  /** Total operations across the definition's buckets. */
+  operations: number;
+  /** Approximate size of the definition's operation history in bytes. */
+  operationBytes: number;
+  /**
+   * Live rows across the definition's buckets, counting a row once per bucket that contains it (the
+   * download-relevant meaning). Sampled estimate for all but tiny definitions (see `rowsEstimated`).
+   */
+  rows: number;
+  /** `operations / max(rows, 1)` across the whole definition. */
+  fragmentation: number;
+  /** True if `rows` (and therefore `fragmentation`) is a sampled estimate rather than an exact count. */
+  rowsEstimated: boolean;
+  /** Suggested maintenance action derived from the operation mix. See {@link suggestBucketAction}. */
+  suggestedAction: BucketAction;
+  /**
+   * Tables making up the (sampled) operation history, ordered by their share of it, largest first. These
+   * are the tables whose rows a defragment should touch.
+   */
+  tables: string[];
+}
+
+export interface BucketReportTotals {
+  /** Number of buckets with stored operations. Estimated when the bucket set was sampled (see `estimated`). */
+  bucketCount: number;
+  /** Sum of operations across all buckets. Estimated when the bucket set was sampled. */
+  operations: number;
+  /** Sum of operation-history bytes across all buckets. Estimated when the bucket set was sampled. */
+  operationBytes: number;
+  /**
+   * True if the totals are estimated because the bucket set was too large to scan in full and was sampled.
+   * Row counts are never totalled here (they are sampled per returned bucket, not across the whole instance).
+   */
+  estimated: boolean;
+}
+
+export interface BucketReport {
+  /** Worst-offender buckets, ranked by operation count then fragmentation. */
+  buckets: BucketStorageStats[];
+  /**
+   * Per-definition rollup, ranked by operation count. Answers "which sync-rules definition should I look
+   * at" where `buckets` answers "which exact buckets". Capped at {@link BUCKET_REPORT_DEFINITION_LIMIT}.
+   */
+  definitions: BucketDefinitionStats[];
+  /** Instance-wide operation totals. Does not include row counts (those are per-bucket estimates only). */
+  totals: BucketReportTotals;
+  /** True if there are more buckets than returned (more than `limit`). */
+  bucketsTruncated: boolean;
+  /**
+   * True if the definition rollup is incomplete: more definitions exist than
+   * {@link BUCKET_REPORT_DEFINITION_LIMIT}, or a definition was dropped because sampling it failed.
+   */
+  definitionsTruncated: boolean;
+}
+
+export interface GetBucketReportOptions {
+  /**
+   * Maximum number of buckets to return, ranked by operation count descending (worst offenders first).
+   * Row counts are sampled per returned bucket, so this also bounds the report's cost. Non-integer or
+   * negative values are floored and clamped to 1. Defaults to {@link DEFAULT_BUCKET_REPORT_LIMIT}.
+   */
+  limit?: number;
+}
+
+/** A bucket's exact operation stats plus its (possibly sampled) row count, before ranking. */
+export interface RankedBucketInput {
+  bucket: string;
+  operations: number;
+  operationBytes: number;
+  rows: number;
+  /**
+   * Operations that carry a row identity (PUT/REMOVE), i.e. everything except compaction residue
+   * (MOVE/CLEAR). Estimated alongside `rows` for sampled buckets.
+   */
+  rowOperations: number;
+  rowsEstimated: boolean;
+  /** Tables in the (sampled) operation history, ordered by their share of it, largest first. */
+  tables: string[];
+}
+
+/** A definition's aggregated operation stats plus its (possibly sampled) row count, before ranking. */
+export interface RankedDefinitionInput {
+  definition: string;
+  bucketCount: number;
+  operations: number;
+  operationBytes: number;
+  rows: number;
+  /** As in {@link RankedBucketInput.rowOperations}, across the whole definition. */
+  rowOperations: number;
+  rowsEstimated: boolean;
+  /** As in {@link RankedBucketInput.tables}, across the whole definition. */
+  tables: string[];
+}
+
+/**
+ * Normalize a requested limit to a positive integer, falling back to {@link DEFAULT_BUCKET_REPORT_LIMIT}.
+ */
+export function resolveBucketReportLimit(limit?: number): number {
+  if (limit == null) {
+    return DEFAULT_BUCKET_REPORT_LIMIT;
+  }
+  return Math.max(1, Math.floor(limit));
+}
+
+/**
+ * Estimate the true distinct row count of a bucket from a random sample of its operations.
+ *
+ * The signal is repetition: a sample that keeps landing on the same rows means few rows, while a sample
+ * where every operation lands on a new row means many. Formally, each operation is included in the sample
+ * with probability `r = sampledOps / operations`, so a row with `k` operations appears with probability
+ * `1 - (1 - r)^k`. Assuming operations are spread roughly evenly across `R` rows (`k = operations / R`),
+ * the expected number of distinct rows in the sample is `R * (1 - (1 - r)^(operations / R))`. That grows
+ * with `R`, so a binary search finds the `R` matching the observed distinct count.
+ *
+ * The naive `distinctRows / r` ignores repetition and over-counts rows (under-stating fragmentation) on
+ * exactly the highly fragmented buckets the report exists to surface.
+ *
+ * Pure (no I/O) so it is unit-testable; storage adapters supply the sampled counts.
+ */
+export function estimateDistinctRows(operations: number, sampledOps: number, distinctRows: number): number {
+  const r = Math.min(1, sampledOps / operations);
+  if (r >= 1) {
+    return distinctRows;
+  }
+  const expectedDistinct = (rows: number) => rows * (1 - Math.pow(1 - r, operations / rows));
+  // The true row count is between the observed distinct count (a lower bound) and one row per operation.
+  // Binary-search that range until it is narrower than a single row, at which point rounding is exact.
+  let lo = distinctRows;
+  let hi = operations;
+  while (hi - lo > 0.5) {
+    const mid = (lo + hi) / 2;
+    if (expectedDistinct(mid) < distinctRows) {
+      lo = mid;
+    } else {
+      hi = mid;
+    }
+  }
+  return Math.round((lo + hi) / 2);
+}
+
+/**
+ * Suggest the maintenance action that reduces what new clients download from a bucket, based on its
+ * operation mix. Grounded in the compaction semantics (see `docs/storage/compacting-operations.md`):
+ *
+ * - A **compact** converts superseded PUT/REMOVE operations into MOVE operations (reclaiming their bytes)
+ *   and collapses a leading run of REMOVE/MOVE operations into one CLEAR. It helps when a bucket carries
+ *   un-compacted superseded history: `rowOperations` well above `rows`.
+ * - A **defragment** (touch every row, then compact) is what collapses the operation count once the history
+ *   is mostly MOVE/CLEAR residue that a compact alone preserves: `operations` well above `rowOperations`.
+ * - When both kinds of overhead are present, or the mix is inconclusive, suggest both.
+ *
+ * The thresholds are heuristics; the report is intended to be re-run after acting on it. Inputs may be
+ * sampled estimates, which is fine at these margins.
+ */
+export function suggestBucketAction(operations: number, rowOperations: number, rows: number): BucketAction {
+  const fragmentation = operations / Math.max(rows, 1);
+  if (fragmentation < BUCKET_ACTION_MIN_FRAGMENTATION) {
+    return 'none';
+  }
+  const residueShare = (operations - rowOperations) / Math.max(operations, 1);
+  const supersededShare = (rowOperations - rows) / Math.max(rowOperations, 1);
+  const defragmentNeeded = residueShare >= BUCKET_ACTION_RESIDUE_SHARE;
+  const compactUseful = supersededShare >= BUCKET_ACTION_SUPERSEDED_SHARE;
+  if (defragmentNeeded && compactUseful) {
+    return 'both';
+  }
+  if (defragmentNeeded) {
+    return 'defragment';
+  }
+  if (compactUseful) {
+    return 'compact';
+  }
+  // Fragmented, but neither share dominates: a mixed history where a compact reclaims part and the rest
+  // needs a defragment.
+  return 'both';
+}
+
+/**
+ * Assemble the final {@link BucketReport} from per-bucket stats, per-definition stats, and instance-wide
+ * totals. Storage adapters select and sample the buckets however is cheapest for them; this owns the shared
+ * fragmentation / ranking / truncation / action logic so it cannot drift. Pure (no I/O) so it is
+ * unit-testable.
+ *
+ * Bucket truncation is derived from the totals; only the adapter knows whether the definition list was cut,
+ * so it passes `definitionsTruncated` in.
+ */
+export function assembleBucketReport(
+  buckets: RankedBucketInput[],
+  definitions: RankedDefinitionInput[],
+  totals: BucketReportTotals,
+  definitionsTruncated = false
+): BucketReport {
+  const stats: BucketStorageStats[] = buckets.map((b) => ({
+    bucket: b.bucket,
+    operations: b.operations,
+    rows: b.rows,
+    operationBytes: b.operationBytes,
+    fragmentation: b.operations / Math.max(b.rows, 1),
+    rowsEstimated: b.rowsEstimated,
+    suggestedAction: suggestBucketAction(b.operations, b.rowOperations, b.rows),
+    tables: b.tables
+  }));
+
+  const definitionStats: BucketDefinitionStats[] = definitions.map((d) => ({
+    definition: d.definition,
+    bucketCount: d.bucketCount,
+    operations: d.operations,
+    operationBytes: d.operationBytes,
+    rows: d.rows,
+    fragmentation: d.operations / Math.max(d.rows, 1),
+    rowsEstimated: d.rowsEstimated,
+    suggestedAction: suggestBucketAction(d.operations, d.rowOperations, d.rows),
+    tables: d.tables
+  }));
+
+  // Worst-first: most operations, then most fragmented.
+  const worstFirst = (a: { operations: number; fragmentation: number }, b: typeof a) =>
+    b.operations - a.operations || b.fragmentation - a.fragmentation;
+  stats.sort(worstFirst);
+  definitionStats.sort(worstFirst);
+
+  return {
+    buckets: stats,
+    definitions: definitionStats,
+    totals,
+    bucketsTruncated: totals.bucketCount > stats.length,
+    definitionsTruncated
+  };
+}
diff --git a/packages/service-core/src/storage/storage-index.ts b/packages/service-core/src/storage/storage-index.ts
index c0cecb5c5..4f4133ddf 100644
--- a/packages/service-core/src/storage/storage-index.ts
+++ b/packages/service-core/src/storage/storage-index.ts
@@ -1,4 +1,5 @@
 export * from './bson.js';
+export * from './bucket-report.js';
 export * from './BucketStorage.js';
 export * from './BucketStorageBatch.js';
 export * from './BucketStorageFactory.js';
diff --git a/packages/service-core/test/src/bucket-report.test.ts b/packages/service-core/test/src/bucket-report.test.ts
new file mode 100644
index 000000000..d31f770be
--- /dev/null
+++ b/packages/service-core/test/src/bucket-report.test.ts
@@ -0,0 +1,231 @@
+import {
+  assembleBucketReport,
+  BucketReportTotals,
+  DEFAULT_BUCKET_REPORT_LIMIT,
+  estimateDistinctRows,
+  RankedBucketInput,
+  RankedDefinitionInput,
+  resolveBucketReportLimit,
+  suggestBucketAction
+} from '@/storage/bucket-report.js';
+import { describe, expect, it } from 'vitest';
+
+// Row-bearing operations default to all operations (no compaction residue) unless overridden.
+const bucket = (
+  name: string,
+  operations: number,
+  rows: number,
+  extra?: Partial<RankedBucketInput>
+): RankedBucketInput => ({
+  bucket: name,
+  operations,
+  rows,
+  operationBytes: extra?.operationBytes ?? 0,
+  rowOperations: extra?.rowOperations ?? operations,
+  rowsEstimated: extra?.rowsEstimated ?? false,
+  tables: extra?.tables ?? []
+});
+
+const definition = (
+  name: string,
+  operations: number,
+  rows: number,
+  extra?: Partial<RankedDefinitionInput>
+): RankedDefinitionInput => ({
+  definition: name,
+  bucketCount: extra?.bucketCount ?? 1,
+  operations,
+  rows,
+  operationBytes: extra?.operationBytes ?? 0,
+  rowOperations: extra?.rowOperations ?? operations,
+  rowsEstimated: extra?.rowsEstimated ?? false,
+  tables: extra?.tables ?? []
+});
+
+const totals = (bucketCount: number, extra?: Partial<BucketReportTotals>): BucketReportTotals => ({
+  bucketCount,
+  operations: extra?.operations ?? 0,
+  operationBytes: extra?.operationBytes ?? 0,
+  estimated: extra?.estimated ?? false
+});
+
+describe('assembleBucketReport', () => {
+  it('derives fragmentation and passes through rowsEstimated and tables', () => {
+    const report = assembleBucketReport(
+      [
+        bucket('global[]', 100, 10, { operationBytes: 1024, tables: ['todos', 'lists'] }),
+        bucket('by_user["u1"]', 30, 30, { rowsEstimated: true })
+      ],
+      [],
+      totals(2)
+    );
+
+    expect(report.buckets.find((b) => b.bucket === 'global[]')).toMatchObject({
+      operations: 100,
+      rows: 10,
+      operationBytes: 1024,
+      fragmentation: 10,
+      rowsEstimated: false,
+      tables: ['todos', 'lists']
+    });
+    expect(report.buckets.find((b) => b.bucket === 'by_user["u1"]')).toMatchObject({
+      fragmentation: 1,
+      rowsEstimated: true
+    });
+  });
+
+  it('ranks buckets worst-first by operations then fragmentation', () => {
+    const report = assembleBucketReport(
+      [bucket('a[]', 5, 5), bucket('b[]', 50, 5), bucket('c[]', 50, 50)],
+      [],
+      totals(3)
+    );
+
+    // b and c both have 50 ops; b is more fragmented (10 vs 1) so it ranks first.
+    expect(report.buckets.map((b) => b.bucket)).toEqual(['b[]', 'c[]', 'a[]']);
+  });
+
+  it('floors rows at 1 so a bucket with operations but no rows is fully fragmented', () => {
+    const report = assembleBucketReport([bucket('gone[]', 42, 0)], [], totals(1));
+
+    expect(report.buckets[0]).toMatchObject({ operations: 42, rows: 0, fragmentation: 42 });
+  });
+
+  it('marks the bucket list truncated when there are more buckets than returned', () => {
+    const truncated = assembleBucketReport([bucket('a[]', 10, 1), bucket('b[]', 5, 1)], [], totals(5));
+    expect(truncated.bucketsTruncated).toBe(true);
+
+    const complete = assembleBucketReport([bucket('a[]', 10, 1), bucket('b[]', 5, 1)], [], totals(2));
+    expect(complete.bucketsTruncated).toBe(false);
+  });
+
+  it('passes the definition truncation flag through, defaulting to complete', () => {
+    expect(assembleBucketReport([], [definition('a', 1, 1)], totals(1)).definitionsTruncated).toBe(false);
+    expect(assembleBucketReport([], [definition('a', 1, 1)], totals(1), true).definitionsTruncated).toBe(true);
+  });
+
+  it('carries the totals through unchanged', () => {
+    const t = totals(2, { operations: 120, operationBytes: 15, estimated: true });
+    const report = assembleBucketReport([bucket('a[]', 100, 4), bucket('b[]', 20, 2)], [], t);
+
+    expect(report.totals).toEqual({ bucketCount: 2, operations: 120, operationBytes: 15, estimated: true });
+  });
+
+  it('assembles and ranks the definition rollup with derived fragmentation and action', () => {
+    const report = assembleBucketReport(
+      [],
+      [
+        definition('1#by_user', 100, 100, { bucketCount: 10 }),
+        definition('1#by_org', 500, 100, { bucketCount: 5, operationBytes: 2048 })
+      ],
+      totals(15)
+    );
+
+    // Ranked by operations: by_org (500) before by_user (100).
+    expect(report.definitions.map((d) => d.definition)).toEqual(['1#by_org', '1#by_user']);
+    expect(report.definitions[0]).toMatchObject({
+      definition: '1#by_org',
+      bucketCount: 5,
+      operations: 500,
+      operationBytes: 2048,
+      rows: 100,
+      fragmentation: 5,
+      suggestedAction: 'compact'
+    });
+    expect(report.definitions[1]).toMatchObject({ fragmentation: 1, suggestedAction: 'none' });
+  });
+
+  it('derives per-bucket suggested actions from the operation mix', () => {
+    const report = assembleBucketReport(
+      [
+        // Healthy: one op per row.
+        bucket('healthy[]', 100, 100),
+        // Un-compacted churn: every op carries a row identity, far more ops than rows.
+        bucket('churned[]', 1000, 100),
+        // Compacted residue: mostly MOVE/CLEAR ops left behind by a compact.
+        bucket('compacted[]', 1000, 100, { rowOperations: 150 })
+      ],
+      [],
+      totals(3)
+    );
+
+    const action = (name: string) => report.buckets.find((b) => b.bucket === name)?.suggestedAction;
+    expect(action('healthy[]')).toEqual('none');
+    expect(action('churned[]')).toEqual('compact');
+    expect(action('compacted[]')).toEqual('defragment');
+  });
+});
+
+describe('suggestBucketAction', () => {
+  it('suggests nothing for healthy buckets', () => {
+    expect(suggestBucketAction(100, 100, 100)).toEqual('none');
+    expect(suggestBucketAction(150, 150, 100)).toEqual('none');
+    expect(suggestBucketAction(0, 0, 0)).toEqual('none');
+  });
+
+  it('suggests compact for un-compacted superseded history', () => {
+    // All operations carry row identity, but there are 10x more of them than rows.
+    expect(suggestBucketAction(1000, 1000, 100)).toEqual('compact');
+  });
+
+  it('suggests defragment when compaction residue dominates', () => {
+    // 850 of 1000 ops are MOVE/CLEAR: a compact already ran and cannot reclaim more.
+    expect(suggestBucketAction(1000, 150, 100)).toEqual('defragment');
+    // A bucket of only MOVE/CLEAR ops (rows 0) is pure residue.
+    expect(suggestBucketAction(500, 0, 0)).toEqual('defragment');
+  });
+
+  it('suggests both when residue and fresh superseded history are both present', () => {
+    // 600 residue ops plus 400 row-bearing ops over 100 rows: defragment for the residue, compact for the churn.
+    expect(suggestBucketAction(1000, 400, 100)).toEqual('both');
+  });
+
+  it('suggests both for a fragmented but inconclusive mix', () => {
+    // Fragmented (frag 2.5), yet neither residue (40%) nor superseded share (33%) dominates.
+    expect(suggestBucketAction(1000, 600, 400)).toEqual('both');
+  });
+});
+
+describe('estimateDistinctRows', () => {
+  it('returns the observed distinct count when the whole bucket was sampled', () => {
+    // r >= 1: nothing was left out, so the observed distinct count is already exact.
+    expect(estimateDistinctRows(100, 100, 40)).toBe(40);
+    expect(estimateDistinctRows(100, 150, 40)).toBe(40);
+  });
+
+  it('recovers a heavily fragmented bucket the naive estimate would inflate', () => {
+    // 10 rows x 1000 ops each; a 10% sample sees ~1000 ops but still only the same 10 distinct rows.
+    // Naive distinct/rate would report 10 / 0.1 = 100 rows (10x too many, so 10x too little fragmentation).
+    const rows = estimateDistinctRows(10_000, 1_000, 10);
+    expect(rows).toBeGreaterThanOrEqual(9);
+    expect(rows).toBeLessThanOrEqual(12);
+  });
+
+  it('recovers a moderately fragmented bucket', () => {
+    // 500 rows x 2 ops each, 50% sample. Ground truth: 500*(1-0.5^2) = 375 distinct sampled rows.
+    // Naive distinct/rate would report 375 / 0.5 = 750 rows; the estimator should recover ~500.
+    const rows = estimateDistinctRows(1_000, 500, 375);
+    expect(rows).toBeGreaterThan(480);
+    expect(rows).toBeLessThan(520);
+  });
+
+  it('matches the naive estimate when there are no sampling collisions', () => {
+    // 2000 rows, 1 op each, 50% sample: no row is seen twice, so distinct/rate is already correct (~2000).
+    const rows = estimateDistinctRows(2_000, 1_000, 1_000);
+    expect(rows).toBeGreaterThan(1_900);
+    expect(rows).toBeLessThan(2_100);
+  });
+});
+
+describe('resolveBucketReportLimit', () => {
+  it('defaults when no limit is given', () => {
+    expect(resolveBucketReportLimit(undefined)).toBe(DEFAULT_BUCKET_REPORT_LIMIT);
+  });
+
+  it('floors and clamps to a positive integer', () => {
+    expect(resolveBucketReportLimit(2.7)).toBe(2);
+    expect(resolveBucketReportLimit(-5)).toBe(1);
+    expect(resolveBucketReportLimit(0)).toBe(1);
+    expect(resolveBucketReportLimit(20)).toBe(20);
+  });
+});
diff --git a/packages/service-core/test/src/routes/admin.test.ts b/packages/service-core/test/src/routes/admin.test.ts
index 8f0fa6e24..f7eea90c9 100644
--- a/packages/service-core/test/src/routes/admin.test.ts
+++ b/packages/service-core/test/src/routes/admin.test.ts
@@ -2,7 +2,7 @@ import { BasicRouterRequest, Context, JwtPayload, ParsedSyncConfigSet, storage }
 import { logger } from '@powersync/lib-services-framework';
 import { SqlSyncRules } from '@powersync/service-sync-rules';
 import { describe, expect, it, vi } from 'vitest';
-import { diagnostics, reprocess, validate } from '../../../src/routes/endpoints/admin.js';
+import { bucketReport, diagnostics, reprocess, validate } from '../../../src/routes/endpoints/admin.js';
 import { mockServiceContext } from './mocks.js';
 
 describe('admin routes', () => {
@@ -209,4 +209,119 @@ bucket_definitions:
       expect(activeBucketStorage.updateSyncRules).not.toHaveBeenCalled();
     });
   });
+
+  describe('bucket-report', () => {
+    const report = {
+      buckets: [
+        {
+          bucket: '1#by_user["u1"]',
+          operations: 4750,
+          rows: 95,
+          operationBytes: 1216000,
+          fragmentation: 50,
+          rowsEstimated: true,
+          suggestedAction: 'compact',
+          tables: ['todos']
+        },
+        {
+          bucket: '1#global[]',
+          operations: 1000,
+          rows: 1000,
+          operationBytes: 3145728,
+          fragmentation: 1,
+          rowsEstimated: false,
+          suggestedAction: 'none',
+          tables: ['lists']
+        }
+      ],
+      definitions: [
+        {
+          definition: '1#by_user',
+          bucketCount: 1,
+          operations: 4750,
+          operationBytes: 1216000,
+          rows: 95,
+          fragmentation: 50,
+          rowsEstimated: true,
+          suggestedAction: 'compact',
+          tables: ['todos']
+        }
+      ],
+      totals: { bucketCount: 2, operations: 5750, operationBytes: 4361728, estimated: false },
+      bucketsTruncated: false,
+      definitionsTruncated: true
+    };
+
+    it('returns the report, forwards the limit, and maps fields to snake_case', async () => {
+      const getBucketReport = vi.fn(async () => report);
+      const activeBucketStorage = {
+        getActiveSyncConfig: vi.fn(async () => ({
+          content: makeSyncConfigContent({}),
+          replicationStream: {},
+          storage: { getBucketReport }
+        }))
+      };
+
+      const response = await bucketReport.handler({
+        context: makeContext(activeBucketStorage),
+        params: { limit: 20 },
+        request
+      });
+
+      expect(getBucketReport).toHaveBeenCalledWith({ limit: 20 });
+      expect(response.buckets[0]).toEqual({
+        bucket: '1#by_user["u1"]',
+        operations: 4750,
+        rows: 95,
+        operation_bytes: 1216000,
+        fragmentation: 50,
+        rows_estimated: true,
+        suggested_action: 'compact',
+        tables: ['todos']
+      });
+      expect(response.definitions).toEqual([
+        {
+          definition: '1#by_user',
+          bucket_count: 1,
+          operations: 4750,
+          operation_bytes: 1216000,
+          rows: 95,
+          fragmentation: 50,
+          rows_estimated: true,
+          suggested_action: 'compact',
+          tables: ['todos']
+        }
+      ]);
+      expect(response.totals).toEqual({
+        bucket_count: 2,
+        operations: 5750,
+        operation_bytes: 4361728,
+        estimated: false
+      });
+      expect(response.buckets_truncated).toBe(false);
+      expect(response.definitions_truncated).toBe(true);
+    });
+
+    it('rejects when there is no active sync config', async () => {
+      const activeBucketStorage = { getActiveSyncConfig: vi.fn(async () => null) };
+
+      await expect(
+        bucketReport.handler({ context: makeContext(activeBucketStorage), params: {}, request })
+      ).rejects.toMatchObject({ errorData: { status: 422, code: 'PSYNC_S4104' } });
+    });
+
+    it('rejects when the storage provider does not support bucket reporting', async () => {
+      const activeBucketStorage = {
+        getActiveSyncConfig: vi.fn(async () => ({
+          content: makeSyncConfigContent({}),
+          replicationStream: {},
+          storage: {}
+        }))
+      };
+
+      await expect(
+        bucketReport.handler({ context: makeContext(activeBucketStorage), params: {}, request })
+      ).rejects.toMatchObject({ errorData: { status: 422, code: 'PSYNC_S2001' } });
+    });
+  });
 });
diff --git a/packages/types/src/routes.ts b/packages/types/src/routes.ts
index 5177e3c77..6036d536b 100644
--- a/packages/types/src/routes.ts
+++ b/packages/types/src/routes.ts
@@ -77,3 +77,97 @@ export type ValidateRequest = t.Encoded<typeof ValidateRequest>;
 
 export const ValidateResponse = SyncRulesStatus;
 export type ValidateResponse = t.Encoded<typeof ValidateResponse>;
+
+export const BucketReportRequest = t.object({
+  /**
+   * Maximum number of buckets to return, ranked by operation count descending (worst offenders first).
+   * Row counts are sampled per returned bucket, so this also bounds the report's cost. Defaults to 50 when
+   * omitted; non-integer or negative values are floored and clamped to 1.
+   */
+  limit: t.number.optional()
+});
+export type BucketReportRequest = t.Encoded<typeof BucketReportRequest>;
+
+export const SuggestedBucketAction = t
+  .literal('none')
+  .or(t.literal('compact'))
+  .or(t.literal('defragment'))
+  .or(t.literal('both'));
+export type SuggestedBucketAction = t.Encoded<typeof SuggestedBucketAction>;
+
+export const BucketStorageStats = t.object({
+  /** Full bucket name, e.g. `by_user["u1"]`. */
+  bucket: t.string,
+  /** Total operations in the bucket's history (PUT/REMOVE/MOVE/CLEAR). */
+  operations: t.number,
+  /** Live rows in the bucket. Exact for small buckets, otherwise a sampled estimate (see `rows_estimated`). */
+  rows: t.number,
+  /** Approximate size of the operation history in bytes. */
+  operation_bytes: t.number,
+  /**
+   * `operations / max(rows, 1)`. ~1 is healthy (fully compacted); higher means more operation-history
+   * overhead that a compact/defragment can reclaim.
+   */
+  fragmentation: t.number,
+  /** True if `rows` (and therefore `fragmentation`) is a sampled estimate rather than an exact count. */
+  rows_estimated: t.boolean,
+  /**
+   * Suggested maintenance action, derived from the bucket's operation mix: `none` (healthy), `compact`
+   * (un-compacted superseded history to reclaim), `defragment` (mostly compaction residue that only a
+   * defragment collapses), or `both`.
+   */
+  suggested_action: SuggestedBucketAction,
+  /**
+   * Tables making up the (sampled) operation history, ordered by their share of it, largest first. These
+   * are the tables whose rows a defragment should touch.
+   */
+  tables: t.array(t.string)
+});
+export type BucketStorageStats = t.Encoded<typeof BucketStorageStats>;
+
+export const BucketDefinitionStats = t.object({
+  /** Definition name as it prefixes bucket names, e.g. `1#by_user` (versioned in storage v2 and later). */
+  definition: t.string,
+  /** Number of buckets in this definition with stored operations. */
+  bucket_count: t.number,
+  /** Total operations across the definition's buckets. */
+  operations: t.number,
+  /** Approximate size of the definition's operation history in bytes. */
+  operation_bytes: t.number,
+  /**
+   * Live rows across the definition's buckets, counting a row once per bucket that contains it. A sampled
+   * estimate for all but tiny definitions (see `rows_estimated`).
+   */
+  rows: t.number,
+  /** `operations / max(rows, 1)` across the whole definition. */
+  fragmentation: t.number,
+  /** True if `rows` (and therefore `fragmentation`) is a sampled estimate rather than an exact count. */
+  rows_estimated: t.boolean,
+  /** Suggested maintenance action for the definition; same values as `buckets[].suggested_action`. */
+  suggested_action: SuggestedBucketAction,
+  /** Tables in the definition's (sampled) operation history, ordered by their share of it, largest first. */
+  tables: t.array(t.string)
+});
+export type BucketDefinitionStats = t.Encoded<typeof BucketDefinitionStats>;
+
+export const BucketReportResponse = t.object({
+  /** Worst-offender buckets, ranked by operation count then fragmentation. */
+  buckets: t.array(BucketStorageStats),
+  /** Per-definition rollup, ranked by operation count then fragmentation. */
+  definitions: t.array(BucketDefinitionStats),
+  totals: t.object({
+    /** Number of buckets with stored operations. Estimated when the bucket set was sampled. */
+    bucket_count: t.number,
+    /** Sum of operations across all buckets. Estimated when the bucket set was sampled. */
+    operations: t.number,
+    /** Sum of operation-history bytes across all buckets. Estimated when the bucket set was sampled. */
+    operation_bytes: t.number,
+    /** True if the totals are estimated because the bucket set was sampled rather than fully scanned. */
+    estimated: t.boolean
+  }),
+  /** True if there are more buckets than returned (more than `limit`). */
+  buckets_truncated: t.boolean,
+  /** True if the definition rollup is incomplete: more definitions exist than the report caps at. */
+  definitions_truncated: t.boolean
+});
+export type BucketReportResponse = t.Encoded<typeof BucketReportResponse>;