diff --git a/DESIGN.md b/DESIGN.md
index e53fceca..49ff3069 100644
--- a/DESIGN.md
+++ b/DESIGN.md
@@ -494,7 +494,11 @@ Do not add motion for decoration. New animation must honor
names, counts, progress percentages or addition/deletion colors.
- Replace skeletons as data arrives, without a minimum display delay. Stored
manifest loading uses the same diff-frame skeleton without unmounting Pierre.
- Background Sync preserves the existing review and usable controls. Errors and
+ Open available stored snapshots while load-time Sync continues; use skeletons
+ only for unavailable data. Replace each file preview independently as it arrives.
+ Keep the header busy indicator active through collection and preview loading.
+ Background Sync preserves the existing review and usable controls; newer heads
+ use the existing “Newer snapshot” notice and explicit “View latest” action. Errors and
empty results replace placeholders with their existing explanation and actions.
## Responsive design
diff --git a/apps/web/src/Header.tsx b/apps/web/src/Header.tsx
index dfdc5bf3..bed00573 100644
--- a/apps/web/src/Header.tsx
+++ b/apps/web/src/Header.tsx
@@ -1,4 +1,9 @@
-import { FolderSyncIcon, ListTreeIcon, RefreshCwIcon } from "lucide-react";
+import {
+ FolderSyncIcon,
+ ListTreeIcon,
+ LoaderCircleIcon,
+ RefreshCwIcon,
+} from "lucide-react";
import { Button } from "@/components/ui/button";
import { AccountButton } from "./account.tsx";
import { ThemeButton } from "./ThemeButton.tsx";
@@ -74,10 +79,20 @@ export function Header() {
size="icon"
aria-label="Sync all checkouts"
title="Sync all checkouts (Alt+R)"
- aria-busy={collecting}
+ aria-busy={collecting || diff.busy}
onClick={refresh}
>
- | null>(
+ null,
+ );
const selectedRef = useRef(selectedId);
useEffect(() => {
selectedRef.current = selectedId;
@@ -213,28 +217,33 @@ export function SessionProvider({ children }: { children: ReactNode }) {
[show],
);
const { remove: removeHead, retry: reloadHeads } = heads;
- // Syncs the local checkouts; null when it fails or one is running.
+ // Load-time navigation joins an active collection. Explicit Sync ignores
+ // duplicate requests while the current collection is running.
const collectLatest = useCallback(
- async (initial = false) => {
- if (collectingRef.current) return null;
- collectingRef.current = true;
+ (initial = false) => {
+ if (collectionRequest.current)
+ return initial ? collectionRequest.current : Promise.resolve(null);
setCollecting(true);
setCollectError("");
- try {
- const { data, error } = await api.POST("/api/v2/collections", {
- params: { query: { initial } },
- });
- if (!data)
- throw new Error(errorDetail(error, "Unable to sync changes"));
- reloadHeads();
- return data;
- } catch (error) {
- setCollectError(errorDetail(error, "Unable to sync changes"));
- return null;
- } finally {
- collectingRef.current = false;
- setCollecting(false);
- }
+ const request = (async () => {
+ try {
+ const { data, error } = await api.POST("/api/v2/collections", {
+ params: { query: { initial } },
+ });
+ if (!data)
+ throw new Error(errorDetail(error, "Unable to sync changes"));
+ reloadHeads();
+ return data;
+ } catch (error) {
+ setCollectError(errorDetail(error, "Unable to sync changes"));
+ return null;
+ } finally {
+ collectionRequest.current = null;
+ setCollecting(false);
+ }
+ })();
+ collectionRequest.current = request;
+ return request;
},
[reloadHeads],
);
@@ -299,58 +308,53 @@ export function SessionProvider({ children }: { children: ReactNode }) {
window.addEventListener("popstate", handleNavigation);
return () => window.removeEventListener("popstate", handleNavigation);
}, []);
- // Live routes resolve once per load or navigation. Local checkouts sync
- // first and open the stream's synced head (or the newest changed one for the
- // root URL); otherwise the catalog's head opens. Later arrivals only raise
- // the newer-head notice. The route whose load-time sync started, and one
- // whose sync failed or synced no matching head.
+ // Open available catalog data while local collection continues. Once selected,
+ // a snapshot stays open; collection only announces newer heads.
const collected = useRef(null);
- const [uncollected, setUncollected] = useState(null);
+ const [collectionResult, setCollectionResult] = useState<{
+ route: Route;
+ contexts: ApiContext[];
+ } | null>(null);
const resolving = route.kind !== "pinned" && !selectedId && !missing;
- const collectOnLoad = resolving && collectEnabled && uncollected !== route;
+ const collectionFinished = collectionResult?.route === route;
+ const collectOnLoad = resolving && collectEnabled && !collectionFinished;
useEffect(() => {
if (!collectOnLoad || collected.current === route) return;
- // Effect replay must not treat the in-flight collection as a failed one.
+ // Effect replay must not start another collection. Completion only publishes
+ // a result; route resolution below cannot replace a user-selected review.
collected.current = route;
void collectLatest(true).then((result) => {
if (result?.failures.length)
setCollectError(
`Unable to sync ${result.failures.length} ${result.failures.length === 1 ? "checkout" : "checkouts"}: ${result.failures.map((failure) => `${failure.path} (${failure.detail})`).join(", ")}`,
);
- const contexts = result?.contexts ?? [];
- const head =
- route.kind === "live"
- ? contexts.find((context) => context.streamId === route.stream)
- : latestChangedContext(contexts);
- if (head)
- openStream(head, {
- replace: true,
- restoreFocus: false,
- preserveNavigation: true,
- });
- else setUncollected(route);
+ setCollectionResult({ route, contexts: result?.contexts ?? [] });
});
- }, [collectOnLoad, route, collectLatest, openStream]);
+ }, [collectOnLoad, route, collectLatest]);
useEffect(() => {
- if (!resolving || !heads.loaded) return;
- if (collectEnabled && uncollected !== route) return;
- const head =
+ if (!resolving) return;
+ const findHead = (contexts: ApiContext[]) =>
route.kind === "live"
- ? heads.contexts.find((context) => context.streamId === route.stream)
- : latestChangedContext(heads.contexts);
+ ? contexts.find((context) => context.streamId === route.stream)
+ : latestChangedContext(contexts);
+ const head =
+ (collectionResult?.route === route
+ ? findHead(collectionResult.contexts)
+ : undefined) ?? findHead(heads.contexts);
if (head)
openStream(head, {
replace: true,
restoreFocus: false,
preserveNavigation: true,
});
- else if (route.kind === "live") setMissing(true);
+ else if (route.kind === "live" && heads.loaded && !collectOnLoad)
+ setMissing(true);
}, [
resolving,
heads.loaded,
heads.contexts,
- collectEnabled,
- uncollected,
+ collectOnLoad,
+ collectionResult,
route,
openStream,
]);
@@ -464,6 +468,23 @@ export function SessionProvider({ children }: { children: ReactNode }) {
)}
+ {collecting && (
+
+
+ )}
{compact &&
}
@@ -522,7 +543,7 @@ export function SessionProvider({ children }: { children: ReactNode }) {
collected.current = null;
setCollectError("");
setMissing(false);
- setUncollected(null);
+ setCollectionResult(null);
}
reloadHeads();
setAttempt((current) => current + 1);
diff --git a/apps/web/src/style.css b/apps/web/src/style.css
index 084cc263..58fe6197 100644
--- a/apps/web/src/style.css
+++ b/apps/web/src/style.css
@@ -52,7 +52,8 @@ button[aria-busy="true"] {
opacity: 0.7;
}
@media (prefers-reduced-motion: no-preference) {
- #refresh[aria-busy="true"] svg {
+ #refresh[aria-busy="true"] svg,
+ #sync[aria-busy="true"] svg {
animation: refresh-spin 800ms linear infinite;
}
}
diff --git a/apps/web/src/use-diff.ts b/apps/web/src/use-diff.ts
index e205985b..9b83da7b 100644
--- a/apps/web/src/use-diff.ts
+++ b/apps/web/src/use-diff.ts
@@ -93,19 +93,24 @@ export function useDiff(contextId: string, mode: DiffMode) {
pending.push(file);
}
}
- setState({
- repository: {
- ...data,
- files: data.files.map((file) => ({ ...file })),
- },
- items: new Map(items),
- busy: true,
- notice: "",
- connected: true,
- });
let next = 0;
let failed = 0;
let loadedBytes = 0;
+ function publish(busy: boolean) {
+ setState({
+ repository: {
+ ...data,
+ files: data.files.map((file) => ({ ...file })),
+ },
+ items: new Map(items),
+ busy,
+ notice: failed
+ ? `${failed} file previews could not load. Refresh to retry.`
+ : "",
+ connected: true,
+ });
+ }
+ publish(true);
async function loadNext() {
while (next < pending.length && !signal.aborted) {
const file = pending[next++];
@@ -218,19 +223,12 @@ export function useDiff(contextId: string, mode: DiffMode) {
),
);
}
+ publish(true);
}
}
await Promise.all(Array.from({ length: 4 }, loadNext));
if (signal.aborted || disposed) return;
- setState({
- repository: data,
- items,
- busy: false,
- connected: true,
- notice: failed
- ? `${failed} file previews could not load. Refresh to retry.`
- : "",
- });
+ publish(false);
} catch (error) {
if (signal.aborted || disposed) return;
setState((previous) => ({
diff --git a/apps/web/test/browser/local-workspace.spec.ts b/apps/web/test/browser/local-workspace.spec.ts
index e3c8b42a..9b90b137 100644
--- a/apps/web/test/browser/local-workspace.spec.ts
+++ b/apps/web/test/browser/local-workspace.spec.ts
@@ -1810,6 +1810,177 @@ test("local first launch waits for collection with an empty catalog", async ({
expect(reviewBounds.width).toBeCloseTo(loadingBounds.width, 0);
});
+for (const initialHeads of ["stored", "arriving"]) {
+ test(`local ${initialHeads} snapshots open during collection and preserve later navigation`, async ({
+ page,
+ }) => {
+ const state = await localWorkspace(page);
+ await page.addInitScript(() => {
+ class IngestionEvents extends EventTarget {
+ notify = () =>
+ this.dispatchEvent(new MessageEvent("ingestion", { data: "{}" }));
+ constructor() {
+ super();
+ window.addEventListener("test-ingestion", this.notify);
+ }
+ close() {
+ window.removeEventListener("test-ingestion", this.notify);
+ }
+ }
+ Object.defineProperty(window, "EventSource", { value: IngestionEvents });
+ });
+ await page.route("**/api/v2/bootstrap", (route) =>
+ route.fulfill({
+ json: {
+ mode: "local",
+ signupEnabled: false,
+ user: null,
+ csrfToken: null,
+ collect: true,
+ },
+ }),
+ );
+ const snapshot = {
+ ...state.context,
+ streamId: "local-stream",
+ latestContextId: state.context.id,
+ };
+ const pinned = { ...state.context, id: "pinned", name: "Pinned review" };
+ let heads = initialHeads === "stored" ? [snapshot] : [];
+ await page.route("**/api/v2/contexts?*", (route) =>
+ route.fulfill({ json: { contexts: heads, nextCursor: null } }),
+ );
+ const opened: string[] = [];
+ await page.route("**/api/v2/contexts/*", (route) => {
+ const id =
+ new URL(route.request().url()).pathname.split("/").at(-1) ?? "";
+ opened.push(id);
+ return route.fulfill({ json: id === "pinned" ? pinned : snapshot });
+ });
+ let pending: Route | undefined;
+ await page.route("**/api/v2/collections*", (route) => {
+ pending = route;
+ });
+ await page.goto("/streams/local-stream");
+ await expect.poll(() => pending !== undefined).toBe(true);
+ if (initialHeads === "arriving") {
+ await expect(page.locator("main.session-status")).toHaveAttribute(
+ "aria-busy",
+ "true",
+ );
+ heads = [snapshot];
+ await page.evaluate(() =>
+ window.dispatchEvent(new Event("test-ingestion")),
+ );
+ }
+ await ready(page);
+ await expect(page.locator("#sync")).toHaveAttribute("aria-busy", "true");
+ await expect(page.locator(".session-status")).toHaveCount(0);
+ // Same-document navigation must survive the earlier collection completing.
+ await page.evaluate(() => {
+ window.history.pushState(null, "", "/contexts/pinned");
+ window.dispatchEvent(new PopStateEvent("popstate"));
+ });
+ await expect(page.locator(".context-identity-name")).toHaveText(
+ "Pinned review",
+ );
+ await ready(page);
+ if (!pending) throw new Error("Missing initial collection");
+ await pending.fulfill({
+ json: {
+ contexts: [{ ...snapshot, id: "newer", latestContextId: "newer" }],
+ failures: [],
+ },
+ });
+ await expect(page.locator("#sync")).toHaveAttribute("aria-busy", "false");
+ await expect(page).toHaveURL(/\/contexts\/pinned$/);
+ expect(opened).toEqual(["local-review", "pinned"]);
+ await ready(page);
+ });
+}
+
+test("unresolved navigation joins collection without reporting a missing stream", async ({
+ page,
+}) => {
+ const state = await localWorkspace(page);
+ await page.route("**/api/v2/bootstrap", (route) =>
+ route.fulfill({
+ json: {
+ mode: "local",
+ signupEnabled: false,
+ user: null,
+ csrfToken: null,
+ collect: true,
+ },
+ }),
+ );
+ await page.route("**/api/v2/contexts?*", (route) =>
+ route.fulfill({ json: { contexts: [], nextCursor: null } }),
+ );
+ const snapshot = {
+ ...state.context,
+ streamId: "local-stream",
+ latestContextId: state.context.id,
+ };
+ await page.route("**/api/v2/contexts/*", (route) =>
+ route.fulfill({ json: snapshot }),
+ );
+ let collections = 0;
+ let pending: Route | undefined;
+ await page.route("**/api/v2/collections*", (route) => {
+ collections++;
+ pending = route;
+ });
+ await page.goto("/");
+ await expect.poll(() => collections).toBe(1);
+ await page.evaluate(() => {
+ window.history.pushState(null, "", "/streams/local-stream");
+ window.dispatchEvent(new PopStateEvent("popstate"));
+ });
+ const status = page.locator("main.session-status");
+ await expect(status).toHaveAttribute("aria-busy", "true");
+ await expect(status).toContainText("Collecting latest changes");
+ if (!pending) throw new Error("Missing initial collection");
+ await pending.fulfill({ json: { contexts: [snapshot], failures: [] } });
+ await ready(page);
+ await expect(page).toHaveURL(/\/streams\/local-stream$/);
+ await expect(page.locator("#sync")).toHaveAttribute("aria-busy", "false");
+ expect(collections).toBe(1);
+});
+
+test("file previews render independently while other previews are pending", async ({
+ page,
+}) => {
+ const state = await localWorkspace(page);
+ await page.route("**/api/v2/contexts/*/diffs/current?*", (route) => {
+ const repository = state.repository("all");
+ const file = repository.files[0];
+ if (!file) throw new Error("Missing fixture file");
+ return route.fulfill({
+ json: {
+ ...repository,
+ files: [file, { ...file, id: "slow-file", path: "src/slow.ts" }],
+ },
+ });
+ });
+ let pending: Route | undefined;
+ await page.route("**/files/slow-file/patch?*", (route) => {
+ pending = route;
+ });
+ await page.goto("/contexts/local-review");
+ await ready(page);
+ await expect.poll(() => pending !== undefined).toBe(true);
+ await expect(page.locator("#refresh")).toHaveAttribute("aria-busy", "true");
+ await expect(page.getByText("Loading diff…", { exact: true })).toBeVisible();
+ if (!pending) throw new Error("Missing held preview");
+ await pending.fulfill({ json: { patch: "", message: "Slow preview ready" } });
+ await expect(
+ page.getByText("Slow preview ready", { exact: true }),
+ ).toBeVisible();
+ await expect(page.locator("#refresh")).toHaveAttribute("aria-busy", "false");
+ await ready(page);
+});
+
test("stored manifest loading keeps the diff surface mounted until content arrives", async ({
page,
}) => {
@@ -1894,6 +2065,9 @@ test("local startup reports failed checkouts while opening collected changes", a
},
}),
);
+ await page.route("**/api/v2/contexts?*", (route) =>
+ route.fulfill({ json: { contexts: [], nextCursor: null } }),
+ );
await page.route("**/api/v2/collections*", (route) =>
route.fulfill({
json: {
@@ -1929,13 +2103,14 @@ test("local live URLs sync the checkouts on every load", async ({ page }) => {
streamId: "local-stream",
latestContextId: id,
});
+ let heads = [snapshot("stored")];
+ const opened: string[] = [];
await page.route("**/api/v2/contexts?*", (route) =>
- route.fulfill({
- json: { contexts: [snapshot("stored")], nextCursor: null },
- }),
+ route.fulfill({ json: { contexts: heads, nextCursor: null } }),
);
await page.route("**/api/v2/contexts/*", (route) => {
const id = new URL(route.request().url()).pathname.split("/").at(-1) ?? "";
+ opened.push(id);
return route.fulfill({ json: snapshot(id) });
});
let collected = 0;
@@ -1959,25 +2134,30 @@ test("local live URLs sync the checkouts on every load", async ({ page }) => {
lastChangedAt: 100,
lastSubmittedAt: 900,
};
- return route.fulfill({
- json: {
- contexts: [
- { ...snapshot(`collected-${collected}`), lastChangedAt: 200 },
- other,
- ],
- failures: [],
- },
- });
+ heads = [
+ { ...snapshot(`collected-${collected}`), lastChangedAt: 200 },
+ other,
+ ];
+ return route.fulfill({ json: { contexts: heads, failures: [] } });
});
await page.goto("/");
- await expect(
- page.getByRole("heading", { name: "Collecting latest changes" }),
- ).toBeVisible();
- release();
await ready(page);
+ await expect(page.locator("#sync")).toHaveAttribute("aria-busy", "true");
+ await expect(page.locator(".session-status")).toHaveCount(0);
+ await expect.poll(() => state.scopes.length).toBe(1);
+ release();
+ await expect(page.locator("#sync")).toHaveAttribute("aria-busy", "false");
await expect(page).toHaveURL(/\/streams\/local-stream$/);
expect(collected).toBe(1);
+ expect(opened).toEqual(["stored"]);
+ await expect(page.locator(".snapshot-notice")).toContainText(
+ "Newer snapshot",
+ );
+ await page.getByRole("button", { name: "View latest" }).click();
+ await expect.poll(() => opened.at(-1)).toBe("collected-1");
+ await ready(page);
+ expect(collected).toBe(1);
// Collapse state belongs to the stream, so it survives the new snapshot.
await page.locator("#viewer .diff-collapse").first().click();
diff --git a/cmd/diffx/ingestion_test.go b/cmd/diffx/ingestion_test.go
index fa766c25..b36b63f0 100644
--- a/cmd/diffx/ingestion_test.go
+++ b/cmd/diffx/ingestion_test.go
@@ -8,6 +8,8 @@ import (
"path/filepath"
"strings"
"testing"
+
+ "github.com/flexdinesh/diffx/internal/ingestion"
)
func cliRepository(t *testing.T) string {
@@ -36,25 +38,37 @@ func cliRepository(t *testing.T) string {
return root
}
-func TestBareInvocationRejectsNonRepositoryWithoutConsumingStdin(t *testing.T) {
- t.Chdir(t.TempDir())
- t.Setenv("DIFFX_CONFIG_PATH", filepath.Join(t.TempDir(), "config.json"))
+func TestNonRepositoryWorkspaceFailsCollectionWithoutConsumingStdin(t *testing.T) {
+ root := t.TempDir()
+ t.Chdir(root)
stdin, err := os.Open(os.DevNull)
if err != nil {
t.Fatal(err)
}
defer stdin.Close()
- var output bytes.Buffer
- if err := run(t.Context(), nil, stdin, &output, io.Discard); err == nil {
- t.Fatal("accepted non-Git current directory")
+ values := options{directory: ".", sourceID: "test-source", trigger: "manual"}
+ input, err := acquireLocalInput(values, stdin)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if input.Kind != "worktree" || len(input.Paths) != 1 || input.Paths[0] != root || len(input.Raw) != 0 {
+ t.Fatalf("workspace input: %#v", input)
+ }
+ publications := 0
+ _, err = collectWorkspaces(t.Context(), values, input.Paths, io.Discard, func(ingestion.Request) error {
+ publications++
+ return nil
+ })
+ if err == nil || !strings.Contains(err.Error(), "no Git checkouts found") {
+ t.Fatalf("empty workspace collection: %v", err)
+ }
+ if publications != 0 {
+ t.Fatalf("empty workspace published %d observations", publications)
}
position, err := stdin.Seek(0, io.SeekCurrent)
if err != nil || position != 0 {
t.Fatalf("stdin consumed: %d %v", position, err)
}
- if output.Len() != 0 {
- t.Fatalf("unexpected startup: %s", output.String())
- }
}
func TestPipedInputKeepsSubmissionDirectoryWithoutGitIdentity(t *testing.T) {
diff --git a/cmd/diffx/local.go b/cmd/diffx/local.go
index 17dd0f7b..88b51a5c 100644
--- a/cmd/diffx/local.go
+++ b/cmd/diffx/local.go
@@ -46,8 +46,8 @@ func runLocal(ctx context.Context, arguments []string, stdin *os.File, stdout, s
var selected daemon.Input
if input.Kind == "worktree" {
label = strings.Join(input.Paths, ", ")
- collect := func(ctx context.Context) (daemon.Collection, error) {
- return collectWorkspaces(ctx, values, input.Paths, stderr)
+ collect := func(ctx context.Context, publish func(ingestion.Request) error) ([]daemon.CollectionFailure, error) {
+ return collectWorkspaces(ctx, values, input.Paths, stderr, publish)
}
selected = daemon.Input{Collect: collect}
} else {
@@ -108,23 +108,23 @@ func runLocal(ctx context.Context, arguments []string, stdin *os.File, stdout, s
// collectWorkspaces collects every checkout discovered under the selected
// paths. Discovery diagnostics go to stderr; failures stay in the collection.
-func collectWorkspaces(ctx context.Context, values options, paths []string, stderr io.Writer) (daemon.Collection, error) {
+func collectWorkspaces(ctx context.Context, values options, paths []string, stderr io.Writer, publish func(ingestion.Request) error) ([]daemon.CollectionFailure, error) {
settings, err := collectionOptions(values)
if err != nil {
- return daemon.Collection{}, err
+ return nil, err
}
- batch, err := collector.CollectWorkspaces(ctx, paths, values.scanDepth, settings)
+ batch, err := collector.CollectWorkspaces(ctx, paths, values.scanDepth, settings, publish)
for _, diagnostic := range batch.Diagnostics {
fmt.Fprintf(stderr, " diffx: %s\n", diagnostic)
}
if err != nil {
- return daemon.Collection{}, err
+ return nil, err
}
- collection := daemon.Collection{Requests: batch.Requests}
+ failures := make([]daemon.CollectionFailure, 0, len(batch.Failures))
for _, failure := range batch.Failures {
- collection.Failures = append(collection.Failures, daemon.CollectionFailure{Path: failure.Path, Detail: failure.Detail})
+ failures = append(failures, daemon.CollectionFailure{Path: failure.Path, Detail: failure.Detail})
}
- return collection, nil
+ return failures, nil
}
func acquireLocalInput(values options, stdin *os.File) (initialInput, error) {
diff --git a/docs/architecture.md b/docs/architecture.md
index 3cb68d74..154e1f3e 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -26,14 +26,18 @@ stopping at repositories, skipping hidden directories) and collects all of their
registered worktrees, because one review process serves a developer's whole
workspace. Stored observations cannot refresh from Git. A local checkout process
discovers and collects again only on user request (`POST /api/v2/collections`):
-opening a live URL or pressing Sync. Each checkout's collection creates a new
-observation independently; one failure is reported without blocking the others.
+opening a live URL or pressing Sync. Discovery feeds checkouts directly to four
+collection workers through a bounded pipeline. Each finished checkout publishes
+immediately through one serial publication owner, while slower Git reads continue.
+Each observation commits independently; one checkout failure is reported without
+blocking the others. Cancellation stops Git work and joins every worker before
+collection returns.
Synced snapshots never replace the open review, matching remote arrivals. Piped
and fixture processes have no collection route. Local queries are scoped to the
discovered checkouts, or to the capture's submission, because the persistent
home-directory database retains every earlier run. The entry point supplies an
-initially empty selection and widens it before ingesting a newly discovered
-checkout; `contextservice` applies it to listing, counting, resolution and deletion, and out-of-scope
+initially empty selection and widens it immediately before publishing each
+collected checkout; `contextservice` applies it to listing, counting, resolution and deletion, and out-of-scope
contexts are not found. A scoped selection without checkouts matches nothing.
Local listeners and the browser start before checkout discovery and collection. A server-owned task performs the
initial collection; the first live-page collection request joins its result,
diff --git a/docs/development.md b/docs/development.md
index fc23e9d8..7c975d6e 100644
--- a/docs/development.md
+++ b/docs/development.md
@@ -269,15 +269,25 @@ controls remain available when the rail is hidden; the header shows the open
review's identity. The selected context scopes every request. The session owner exposes
local checkout sync when bootstrap reports `collect`; loading `/` or a live
`/streams/{streamId}` URL, the header's “Sync all checkouts” icon and Alt+R call
-`POST /api/v2/collections`. Load-time sync opens the stream's synced head, or the
-most recently active changed one for `/`, and shows a workspace skeleton with
-“Collecting latest changes” in its status bar until it resolves. Skeletons retain
-shell geometry and display preferences, hide decorative placeholders from assistive
+`POST /api/v2/collections`. Load-time sync runs in the background while the
+first available stored head opens: the requested stream's head, or the most
+recently active changed one for `/`. Collection completion never replaces an
+open review; the newer-head notice offers “View latest”. A workspace skeleton
+with “Collecting latest changes” appears only while no review is available.
+Skeletons retain shell geometry and display preferences, hide decorative placeholders from assistive
technology, and respect reduced motion. Stored manifest loading uses the same diff
-skeleton while keeping Pierre mounted. The listener opens before background startup collection. Load-time
-requests send `initial=true` to join startup once without collecting twice;
-explicit Sync requests always collect fresh snapshots. While discovery is pending,
-the process scope matches no stored contexts. Startup failures keep the server
+skeleton while keeping Pierre mounted. File previews replace their placeholders
+individually as they arrive; loaded previews remain usable while others load.
+The header busy indicator stays active through collection and preview loading.
+The listener opens before background startup collection. Collection uses four
+Git workers fed directly by discovery and publishes each completed checkout
+immediately through one serial publisher. Discovery, Git reads and publication
+overlap through bounded channels. Ingestion events refresh the catalog while
+other checkouts continue; the batch response waits for every worker. Cancellation
+and fatal publication errors cancel and join the workers before returning.
+Load-time requests send `initial=true` to join startup once without collecting
+twice; explicit Sync requests always collect fresh snapshots. The process scope
+starts empty and widens with each published checkout, while discovery may continue. Startup failures keep the server
available and show Retry when no stored review can open. An explicit sync never
replaces the open review: it reloads the heads,
so the newer-head notice, drawer dots and global unread action announce new
diff --git a/docs/system.md b/docs/system.md
index b8406f50..c55b1f06 100644
--- a/docs/system.md
+++ b/docs/system.md
@@ -37,9 +37,12 @@ for contracts, durability rules and migration details.
`diffx [PATH]` is one foreground process: serve UI/REST/MCP immediately, then
discover and collect the checkouts under its paths and ingest directly in a
-background task. The UI shows loading feedback while waiting for the first
-snapshots; failures remain visible with Retry. Omitted paths mean the
-current directory; `--path` repeats and `--scan-depth` (default 8) bounds the
+background task. Discovery feeds four workers as checkouts are found; a serial
+publisher ingests each completed checkout immediately, so its snapshot becomes queryable
+while slower checkouts continue. Publication commits and emits the existing
+ingestion event for each observation. The UI opens available snapshots and shows
+loading feedback only for missing data; failures remain visible with Retry.
+Omitted paths mean the current directory; `--path` repeats and `--scan-depth` (default 8) bounds the
search. Captured snapshots remain fixed after edits and branch switches until the
user requests collection by loading a live URL or pressing Sync, which discovers
the paths again and ingests a new snapshot of every checkout; nothing watches Git.
@@ -83,8 +86,14 @@ scan depth, stopping at repositories and skipping hidden, `node_modules` and
found, at most 256 checkouts from 4096 directories. Each checkout gets its own
submission ID. Explicit `--branch` recovery remains available for remote collection.
-Local composition calls `contextservice.Ingest` directly. Remote collectors
-POST to `/api/v2/ingestion-jobs`, receiving 202 with a durable job/status URL.
+Local composition calls `contextservice.Ingest` directly as each checkout finishes
+collection. Discovery feeds up to four concurrent Git reads, while one serial
+publisher commits their completed requests. A slow checkout does not hold already
+collected snapshots; a failed checkout never publishes an empty result. Cancellation or a fatal publication
+failure cancels and joins the workers. The collection HTTP response still reports
+the complete batch and its checkout failures after every worker finishes.
+Remote collectors POST to `/api/v2/ingestion-jobs`, receiving 202 with a durable
+job/status URL.
Acceptance records owner, payload hash, submission identity and arrival sequence
atomically. A leased worker calls the same ingestion operation. The observation,
scopes, session associations, retry identity and stream head commit atomically.
@@ -114,10 +123,13 @@ opens the stream's latest head when loaded or navigated to, while
live URLs; the picker opens pinned ones; piped snapshots have no stream and stay
pinned. A live URL resolves once, so later arrivals raise the notice instead of
swapping the open review. With local collection enabled, loading `/` or a live
-URL first syncs the checkouts, joining startup on the first load through
-`POST /api/v2/collections?initial=true`, then opens the stream's synced head or,
-for `/`, the most recently active changed one; a failed sync opens the stored
-head and reports the error. Otherwise `/` opens the most recently active changed head's
+URL syncs the checkouts in the background, joining startup on the first load
+through `POST /api/v2/collections?initial=true`. The first available catalog
+head opens immediately: the requested stream's head or, for `/`, the most
+recently active changed one. The open snapshot stays selected when collection
+finishes; a newer head raises the “Newer snapshot” notice with “View latest”.
+Without an available head, the UI waits for collection; failures remain visible
+and offer Retry. Otherwise `/` opens the most recently active changed head's
live URL. Startup selection and the local default MCP context use the same
activity ordering.
Repos and Diff share one navigation panel beside the activity rail, aligned
diff --git a/internal/collector/discovery.go b/internal/collector/discovery.go
index 76dbde7a..eaff3e75 100644
--- a/internal/collector/discovery.go
+++ b/internal/collector/discovery.go
@@ -86,12 +86,80 @@ func DiscoverCheckout(ctx context.Context, input, sourceID, base string) (Discov
// including those outside the paths. It never fetches, switches branches,
// creates worktrees or follows symbolic links.
func DiscoverWorkspaces(ctx context.Context, inputs []string, depth int, sourceID, base string) (Discovery, error) {
+ return discoverWorkspaces(ctx, inputs, depth, sourceID, base, nil)
+}
+
+func discoverWorkspaces(ctx context.Context, inputs []string, depth int, sourceID, base string, visit func(DiscoverySource) error) (Discovery, error) {
var result Discovery
if base == "" {
base = "auto"
}
- type repository struct{ input, root string }
- roots := make([]repository, 0)
+ seenRepositories, seenSources := make(map[string]bool), make(map[string]string)
+ sourceLimitReported := false
+ limitReached := func() bool {
+ if len(result.Sources) < checkoutLimit {
+ return false
+ }
+ if !sourceLimitReported {
+ result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("checkout discovery reached %d-checkout limit; narrow --path or --scan-depth", checkoutLimit))
+ sourceLimitReported = true
+ }
+ return true
+ }
+ discoverRepository := func(input, root string) error {
+ if limitReached() {
+ return nil
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ facts, err := diffsource.Metadata(ctx, root)
+ if err != nil {
+ result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("identify repository %s: %v", root, err))
+ return nil
+ }
+ // Enumeration deduplicates physical Git stores only. Independent clones
+ // can share a repository key while retaining distinct checkout identities.
+ if seenRepositories[facts.CommonDir] {
+ return nil
+ }
+ seenRepositories[facts.CommonDir] = true
+ _, worktrees, err := diffsource.RepositoryWorktrees(ctx, root)
+ if err != nil {
+ result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("list worktrees %s: %v", root, err))
+ return nil
+ }
+ for _, worktree := range worktrees {
+ if limitReached() {
+ break
+ }
+ if _, err := os.Stat(worktree.Root); err != nil {
+ result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("worktree unavailable %s: %v", worktree.Root, err))
+ continue
+ }
+ identity, err := SourceIdentity(ctx, worktree.Root, sourceID, "", base)
+ if err != nil {
+ result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("identify %s: %v", worktree.Root, err))
+ continue
+ }
+ if previous, exists := seenSources[identity]; exists {
+ if previous != worktree.Root {
+ result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("collector identity copied across %s and %s; duplicate source skipped, use separate source IDs", previous, worktree.Root))
+ }
+ continue
+ }
+ seenSources[identity] = worktree.Root
+ source := DiscoverySource{InputPath: input, Path: worktree.Root, Base: base, Identity: identity}
+ result.Sources = append(result.Sources, source)
+ if visit != nil {
+ if err := visit(source); err != nil {
+ return err
+ }
+ }
+ }
+ return nil
+ }
+
visited := 0
visitLimitReported := false
var scan func(input, path string, remaining int) error
@@ -119,8 +187,7 @@ func DiscoverWorkspaces(ctx context.Context, inputs []string, depth int, sourceI
// Only a .git entry can start a repository below a non-repository parent.
source, err := diffsource.OpenRepository(ctx, path)
if err == nil {
- roots = append(roots, repository{input, source.Root()})
- return nil
+ return discoverRepository(input, source.Root())
}
if !errors.Is(err, diffsource.ErrNotRepository) {
result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("inspect %s: %v", path, err))
@@ -152,7 +219,9 @@ func DiscoverWorkspaces(ctx context.Context, inputs []string, depth int, sourceI
// An explicit path may sit anywhere inside a repository.
source, err := diffsource.OpenRepository(ctx, input)
if err == nil {
- roots = append(roots, repository{input, source.Root()})
+ if err := discoverRepository(input, source.Root()); err != nil {
+ return result, err
+ }
continue
}
if !errors.Is(err, diffsource.ErrNotRepository) {
@@ -163,63 +232,5 @@ func DiscoverWorkspaces(ctx context.Context, inputs []string, depth int, sourceI
return result, err
}
}
- seenRepositories, seenSources := make(map[string]bool), make(map[string]string)
- sourceLimitReported := false
- limitReached := func() bool {
- if len(result.Sources) < checkoutLimit {
- return false
- }
- if !sourceLimitReported {
- result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("checkout discovery reached %d-checkout limit; narrow --path or --scan-depth", checkoutLimit))
- sourceLimitReported = true
- }
- return true
- }
- for _, found := range roots {
- if limitReached() {
- break
- }
- if err := ctx.Err(); err != nil {
- return result, err
- }
- facts, err := diffsource.Metadata(ctx, found.root)
- if err != nil {
- result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("identify repository %s: %v", found.root, err))
- continue
- }
- // Enumeration deduplicates physical Git stores only. Independent clones
- // can share a repository key while retaining distinct checkout identities.
- if seenRepositories[facts.CommonDir] {
- continue
- }
- seenRepositories[facts.CommonDir] = true
- _, worktrees, err := diffsource.RepositoryWorktrees(ctx, found.root)
- if err != nil {
- result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("list worktrees %s: %v", found.root, err))
- continue
- }
- for _, worktree := range worktrees {
- if limitReached() {
- break
- }
- if _, err := os.Stat(worktree.Root); err != nil {
- result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("worktree unavailable %s: %v", worktree.Root, err))
- continue
- }
- identity, err := SourceIdentity(ctx, worktree.Root, sourceID, "", base)
- if err != nil {
- result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("identify %s: %v", worktree.Root, err))
- continue
- }
- if previous, exists := seenSources[identity]; exists {
- if previous != worktree.Root {
- result.Diagnostics = append(result.Diagnostics, fmt.Sprintf("collector identity copied across %s and %s; duplicate source skipped, use separate source IDs", previous, worktree.Root))
- }
- continue
- }
- seenSources[identity] = worktree.Root
- result.Sources = append(result.Sources, DiscoverySource{InputPath: found.input, Path: worktree.Root, Base: base, Identity: identity})
- }
- }
return result, ctx.Err()
}
diff --git a/internal/collector/discovery_test.go b/internal/collector/discovery_test.go
index 0b5d654e..097313d4 100644
--- a/internal/collector/discovery_test.go
+++ b/internal/collector/discovery_test.go
@@ -2,11 +2,20 @@ package collector
import (
"context"
+ "errors"
+ "fmt"
+ "github.com/flexdinesh/diffx/internal/diffsource"
+ "github.com/flexdinesh/diffx/internal/ingestion"
"os"
+ "os/exec"
"path/filepath"
+ "runtime"
"slices"
"strings"
+ "sync"
+ "sync/atomic"
"testing"
+ "time"
)
func committed(t *testing.T, root string) string {
@@ -170,20 +179,22 @@ func TestCollectWorkspacesCollectsEachCheckoutIndependently(t *testing.T) {
write(t, first, "file.txt", "after\n")
write(t, second, "file.txt", "after\n")
git(t, second, "add", ".")
- batch, err := CollectWorkspaces(t.Context(), []string{workspace}, 8, Options{SourceID: "machine", Hostname: "host", SubmissionID: "shared"})
- if err != nil || len(batch.Requests) != 2 || len(batch.Failures) != 0 {
+ var requests []ingestion.Request
+ batch, err := CollectWorkspaces(t.Context(), []string{workspace}, 8, Options{SourceID: "machine", Hostname: "host", SubmissionID: "shared"}, func(input ingestion.Request) error { requests = append(requests, input); return nil })
+ slices.SortFunc(requests, func(a, b ingestion.Request) int { return strings.Compare(a.Metadata.Root, b.Metadata.Root) })
+ if err != nil || len(requests) != 2 || len(batch.Failures) != 0 {
t.Fatalf("batch %#v err %v", batch, err)
}
- if batch.Requests[0].Metadata.Root != first || batch.Requests[1].Metadata.Root != second {
- t.Fatalf("unexpected roots: %s %s", batch.Requests[0].Metadata.Root, batch.Requests[1].Metadata.Root)
+ if requests[0].Metadata.Root != first || requests[1].Metadata.Root != second {
+ t.Fatalf("unexpected roots: %s %s", requests[0].Metadata.Root, requests[1].Metadata.Root)
}
- if batch.Requests[0].SubmissionID == batch.Requests[1].SubmissionID || batch.Requests[0].SubmissionID == "shared" {
+ if requests[0].SubmissionID == requests[1].SubmissionID || requests[0].SubmissionID == "shared" {
t.Fatal("checkouts share a submission ID")
}
- if batch.Requests[0].Metadata.TriggerRoot != workspace {
- t.Fatalf("trigger root %q", batch.Requests[0].Metadata.TriggerRoot)
+ if requests[0].Metadata.TriggerRoot != workspace {
+ t.Fatalf("trigger root %q", requests[0].Metadata.TriggerRoot)
}
- if _, err := CollectWorkspaces(t.Context(), []string{t.TempDir()}, 8, Options{SourceID: "machine", Hostname: "host"}); err == nil || !strings.Contains(err.Error(), "no Git checkouts found") {
+ if _, err := CollectWorkspaces(t.Context(), []string{t.TempDir()}, 8, Options{SourceID: "machine", Hostname: "host"}, func(ingestion.Request) error { return nil }); err == nil || !strings.Contains(err.Error(), "no Git checkouts found") {
t.Fatalf("empty workspace: %v", err)
}
}
@@ -195,11 +206,236 @@ func TestCollectWorkspacesReportsFailedCheckoutWithoutBlockingOthers(t *testing.
write(t, healthy, "file.txt", "after\n")
// A base that cannot resolve fails only the checkout without that ref.
git(t, healthy, "branch", "baseline")
- batch, err := CollectWorkspaces(t.Context(), []string{workspace}, 8, Options{SourceID: "machine", Hostname: "host", Base: "baseline"})
- if err != nil || len(batch.Requests) != 1 || batch.Requests[0].Metadata.Root != healthy {
+ var requests []ingestion.Request
+ batch, err := CollectWorkspaces(t.Context(), []string{workspace}, 8, Options{SourceID: "machine", Hostname: "host", Base: "baseline"}, func(input ingestion.Request) error { requests = append(requests, input); return nil })
+ if err != nil || len(requests) != 1 || requests[0].Metadata.Root != healthy {
t.Fatalf("batch %#v err %v", batch, err)
}
if len(batch.Failures) != 1 || batch.Failures[0].Path != broken || batch.Failures[0].Detail == "" {
t.Fatalf("failures %#v", batch.Failures)
}
}
+
+func TestCollectWorkspacesPublishesBeforeOtherCheckoutFinishes(t *testing.T) {
+ workspace := t.TempDir()
+ for _, name := range []string{"first", "second"} {
+ root := initialized(t, filepath.Join(workspace, name))
+ write(t, root, "file.txt", "after\n")
+ }
+ blocked, release := make(chan struct{}), make(chan struct{})
+ var releaseOnce sync.Once
+ t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) })
+ var observed int
+ options := Options{SourceID: "machine", Hostname: "host", OnSource: func(ingestion.Metadata, diffsource.Comparison) {
+ observed++
+ if observed == 2 {
+ close(blocked)
+ <-release
+ }
+ }}
+ published := make(chan ingestion.Request, 2)
+ done := make(chan error, 1)
+ go func() {
+ _, err := CollectWorkspaces(t.Context(), []string{workspace}, 8, options, func(input ingestion.Request) error {
+ published <- input
+ return nil
+ })
+ done <- err
+ }()
+ select {
+ case <-blocked:
+ case <-time.After(5 * time.Second):
+ t.Fatal("second checkout did not block")
+ }
+ select {
+ case input := <-published:
+ if len(input.Scopes) == 0 || len(input.Scopes[0].Snapshot.Files) != 1 {
+ t.Fatalf("incomplete published checkout: %#v", input)
+ }
+ case <-done:
+ t.Fatal("collection finished while checkout was held")
+ case <-time.After(5 * time.Second):
+ t.Fatal("ready checkout waited for held checkout")
+ }
+ releaseOnce.Do(func() { close(release) })
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("collection did not finish")
+ }
+ if len(published) != 1 {
+ t.Fatal("remaining checkout was not published")
+ }
+}
+
+func TestCollectWorkspacesJoinsWorkersAfterCancellationAndPublishFailure(t *testing.T) {
+ for _, failPublish := range []bool{false, true} {
+ t.Run(fmt.Sprintf("publish-failure-%v", failPublish), func(t *testing.T) {
+ workspace := t.TempDir()
+ for _, name := range []string{"first", "second", "third", "fourth", "fifth"} {
+ root := initialized(t, filepath.Join(workspace, name))
+ write(t, root, "file.txt", "after\n")
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ blocked, release := make(chan struct{}), make(chan struct{})
+ var releaseOnce sync.Once
+ t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) })
+ var observed int
+ var exited atomic.Bool
+ options := Options{SourceID: "machine", OnSource: func(ingestion.Metadata, diffsource.Comparison) {
+ observed++
+ if observed == 2 {
+ close(blocked)
+ <-release
+ exited.Store(true)
+ }
+ }}
+ publishFailure := errors.New("publication failed")
+ published := make(chan struct{}, 1)
+ done := make(chan error, 1)
+ go func() {
+ _, err := CollectWorkspaces(ctx, []string{workspace}, 8, options, func(ingestion.Request) error {
+ <-blocked
+ published <- struct{}{}
+ if failPublish {
+ return publishFailure
+ }
+ return nil
+ })
+ done <- err
+ }()
+ select {
+ case <-blocked:
+ case <-time.After(5 * time.Second):
+ t.Fatal("checkout did not block")
+ }
+ select {
+ case <-published:
+ case <-time.After(5 * time.Second):
+ t.Fatal("first checkout did not publish")
+ }
+ if !failPublish {
+ cancel()
+ }
+ select {
+ case err := <-done:
+ t.Fatalf("returned before worker joined: %v", err)
+ case <-time.After(30 * time.Millisecond):
+ }
+ releaseOnce.Do(func() { close(release) })
+ select {
+ case err := <-done:
+ want := context.Canceled
+ if failPublish {
+ want = publishFailure
+ }
+ if !errors.Is(err, want) {
+ t.Fatalf("collection error %v, want %v", err, want)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("collection did not join workers")
+ }
+ if !exited.Load() {
+ t.Fatal("blocked worker did not exit")
+ }
+ if len(published) != 0 {
+ t.Fatal("published another checkout after stopping")
+ }
+ })
+ }
+}
+
+func TestCollectWorkspacesPublishesBeforeWorkspaceDiscoveryFinishes(t *testing.T) {
+ if runtime.GOOS == "windows" {
+ t.Skip("Git wrapper requires a POSIX shell")
+ }
+ workspace := t.TempDir()
+ first := initialized(t, filepath.Join(workspace, "first"))
+ second := initialized(t, filepath.Join(workspace, "second"))
+ write(t, first, "file.txt", "first change\n")
+ write(t, second, "file.txt", "second change\n")
+ realGit, err := exec.LookPath("git")
+ if err != nil {
+ t.Fatal(err)
+ }
+ bin := t.TempDir()
+ blocked, release := filepath.Join(bin, "blocked"), filepath.Join(bin, "release")
+ script := `#!/bin/sh
+if [ "$PWD" = "$DIFFX_TEST_SLOW_CHECKOUT" ]; then
+ case " $* " in
+ *" rev-parse --show-toplevel "*)
+ printf 'blocked\n' > "$DIFFX_TEST_DISCOVERY_BLOCKED"
+ while [ ! -f "$DIFFX_TEST_DISCOVERY_RELEASE" ]; do sleep 0.01; done
+ ;;
+ esac
+fi
+exec "$DIFFX_TEST_REAL_GIT" "$@"
+`
+ if err := os.WriteFile(filepath.Join(bin, "git"), []byte(script), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ t.Setenv("DIFFX_TEST_REAL_GIT", realGit)
+ t.Setenv("DIFFX_TEST_SLOW_CHECKOUT", second)
+ t.Setenv("DIFFX_TEST_DISCOVERY_BLOCKED", blocked)
+ t.Setenv("DIFFX_TEST_DISCOVERY_RELEASE", release)
+ t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH"))
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ published := make(chan ingestion.Request, 2)
+ done := make(chan error, 1)
+ stopped := make(chan struct{})
+ go func() {
+ defer close(stopped)
+ _, err := CollectWorkspaces(ctx, []string{first, second}, 8, Options{SourceID: "machine"}, func(input ingestion.Request) error {
+ published <- input
+ return nil
+ })
+ done <- err
+ }()
+ t.Cleanup(func() {
+ cancel()
+ _ = os.WriteFile(release, nil, 0o600)
+ select {
+ case <-stopped:
+ case <-time.After(5 * time.Second):
+ t.Error("collection did not stop")
+ }
+ })
+ deadline := time.After(5 * time.Second)
+ for {
+ if _, err := os.Stat(blocked); err == nil {
+ break
+ }
+ select {
+ case <-deadline:
+ t.Fatal("later checkout discovery did not block")
+ case <-time.After(10 * time.Millisecond):
+ }
+ }
+ select {
+ case input := <-published:
+ if input.Metadata.Root != first {
+ t.Fatalf("unexpected first checkout: %s", input.Metadata.Root)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("ready checkout waited for later discovery")
+ }
+ if err := os.WriteFile(release, nil, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("collection did not finish")
+ }
+ if len(published) != 1 {
+ t.Fatal("second checkout was not published")
+ }
+}
diff --git a/internal/collector/workspaces.go b/internal/collector/workspaces.go
index 68759439..2df75536 100644
--- a/internal/collector/workspaces.go
+++ b/internal/collector/workspaces.go
@@ -21,7 +21,6 @@ type Failure struct {
// Batch is one collection of every checkout discovered under the workspaces.
type Batch struct {
- Requests []ingestion.Request
Failures []Failure
Diagnostics []string
}
@@ -29,23 +28,14 @@ type Batch struct {
// CollectWorkspaces discovers the workspaces again and collects each checkout
// independently with its own submission ID. A failed checkout never blocks the
// others or becomes an empty observation. Finding no checkout is an error.
-func CollectWorkspaces(ctx context.Context, paths []string, depth int, options Options) (Batch, error) {
+// publish runs serially as requests finish. Its failure cancels and joins all
+// remaining collection before returning. Four workers and a bounded result
+// buffer limit completed requests waiting to publish.
+func CollectWorkspaces(ctx context.Context, paths []string, depth int, options Options, publish func(ingestion.Request) error) (Batch, error) {
options, err := defaults(options)
if err != nil {
return Batch{}, err
}
- discovery, err := DiscoverWorkspaces(ctx, paths, depth, options.SourceID, options.Base)
- if err != nil {
- return Batch{}, err
- }
- batch := Batch{Diagnostics: discovery.Diagnostics}
- if len(discovery.Sources) == 0 {
- detail := "no Git checkouts found in " + strings.Join(paths, ", ")
- if len(discovery.Diagnostics) > 0 {
- detail += ": " + strings.Join(discovery.Diagnostics, "; ")
- }
- return batch, errors.New(detail)
- }
// Collection runs concurrently; callbacks observe one source at a time.
var observed sync.Mutex
if notify := options.OnSource; notify != nil {
@@ -56,34 +46,87 @@ func CollectWorkspaces(ctx context.Context, paths []string, depth int, options O
}
}
type outcome struct {
+ path string
request ingestion.Request
err error
}
- outcomes := make([]outcome, len(discovery.Sources))
- slots := make(chan struct{}, collectionWorkers)
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+ jobs := make(chan DiscoverySource)
+ outcomes := make(chan outcome, collectionWorkers)
+ type discovered struct {
+ value Discovery
+ err error
+ }
+ discoveryDone := make(chan discovered, 1)
+ go func() {
+ value, err := discoverWorkspaces(ctx, paths, depth, options.SourceID, options.Base, func(source DiscoverySource) error {
+ select {
+ case jobs <- source:
+ return nil
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ })
+ close(jobs)
+ discoveryDone <- discovered{value, err}
+ }()
var workers sync.WaitGroup
- for index, source := range discovery.Sources {
+ for range collectionWorkers {
workers.Add(1)
go func() {
defer workers.Done()
- slots <- struct{}{}
- defer func() { <-slots }()
- settings := options
- settings.SubmissionID, settings.TriggerRoot = uuid.NewString(), source.InputPath
- request, err := Collect(ctx, source.Path, settings)
- outcomes[index] = outcome{request, err}
+ for source := range jobs {
+ if ctx.Err() != nil {
+ return
+ }
+ settings := options
+ settings.SubmissionID, settings.TriggerRoot = uuid.NewString(), source.InputPath
+ request, err := Collect(ctx, source.Path, settings)
+ select {
+ case outcomes <- outcome{source.Path, request, err}:
+ case <-ctx.Done():
+ return
+ }
+ }
}()
}
- workers.Wait()
- if err := ctx.Err(); err != nil {
- return Batch{}, err
- }
- for index, result := range outcomes {
+ go func() {
+ workers.Wait()
+ close(outcomes)
+ }()
+ batch := Batch{}
+ var publishError error
+ for result := range outcomes {
+ if ctx.Err() != nil {
+ continue
+ }
if result.err != nil {
- batch.Failures = append(batch.Failures, Failure{Path: discovery.Sources[index].Path, Detail: result.err.Error()})
+ batch.Failures = append(batch.Failures, Failure{Path: result.path, Detail: result.err.Error()})
continue
}
- batch.Requests = append(batch.Requests, result.request)
+ if err := publish(result.request); err != nil {
+ publishError = err
+ cancel()
+ }
+ }
+ discovery := <-discoveryDone
+ batch.Diagnostics = discovery.value.Diagnostics
+ if publishError != nil {
+ return batch, publishError
+ }
+ if err := ctx.Err(); err != nil {
+ return batch, err
+ }
+ if discovery.err != nil {
+ return batch, discovery.err
+ }
+ if len(discovery.value.Sources) == 0 {
+ detail := "no Git checkouts found in " + strings.Join(paths, ", ")
+ if len(batch.Diagnostics) > 0 {
+ detail += ": " + strings.Join(batch.Diagnostics, "; ")
+ }
+ return batch, errors.New(detail)
}
return batch, nil
}
diff --git a/internal/daemon/foreground.go b/internal/daemon/foreground.go
index 6be0016c..ff9e75be 100644
--- a/internal/daemon/foreground.go
+++ b/internal/daemon/foreground.go
@@ -17,6 +17,7 @@ import (
type (
Collection = serverapp.Collection
CollectionFailure = serverapp.CollectionFailure
+ Collect = serverapp.Collect
Published = serverapp.Published
)
@@ -24,7 +25,7 @@ type (
// background at startup and again on explicit user request.
type Input struct {
Collection Collection
- Collect func(context.Context) (Collection, error)
+ Collect Collect
}
// RunForeground serializes discovery, replacement and publication. Shutdown
diff --git a/internal/daemon/server.go b/internal/daemon/server.go
index ae37181d..fe69dece 100644
--- a/internal/daemon/server.go
+++ b/internal/daemon/server.go
@@ -67,7 +67,7 @@ func Run(ctx context.Context, settings Settings, runtimeDir string, initial *ing
// runServer scopes reads to the capture or discovered checkouts. Checkout
// discovery and collection start after serving, with an initially empty scope.
-func runServer(ctx context.Context, cancel context.CancelFunc, settings Settings, runtimeDir, instanceID string, initial Collection, collect func(context.Context) (Collection, error), ready func(Status, Published), logger *log.Logger) error {
+func runServer(ctx context.Context, cancel context.CancelFunc, settings Settings, runtimeDir, instanceID string, initial Collection, collect Collect, ready func(Status, Published), logger *log.Logger) error {
settings = normalizeSettings(settings)
if settings.RetentionDays < 1 || settings.RetentionDays > 106751 {
return errors.New("retention-days must be between 1 and 106751")
diff --git a/internal/daemon/server_test.go b/internal/daemon/server_test.go
index 58631b28..a8370875 100644
--- a/internal/daemon/server_test.go
+++ b/internal/daemon/server_test.go
@@ -18,6 +18,7 @@ import (
"github.com/flexdinesh/diffx/internal/collector"
"github.com/flexdinesh/diffx/internal/contextservice"
+ "github.com/flexdinesh/diffx/internal/ingestion"
"github.com/flexdinesh/diffx/internal/reviewstore"
)
@@ -58,11 +59,11 @@ func TestCheckoutServerServesBeforeCollectionAndWaitsForCollectorShutdown(t *tes
go func() {
defer close(stopped)
done <- runServer(ctx, cancel, Settings{Host: "127.0.0.1", Port: 0, State: state}, "", "async-test", Collection{},
- func(ctx context.Context) (Collection, error) {
+ func(ctx context.Context, _ func(ingestion.Request) error) ([]CollectionFailure, error) {
<-ctx.Done()
close(collectionCanceled)
<-release
- return Collection{}, ctx.Err()
+ return nil, ctx.Err()
}, func(status Status, published Published) {
if len(published.Submissions) != 0 {
t.Error("ready must not wait for checkout collection")
diff --git a/internal/serverapp/capabilities_test.go b/internal/serverapp/capabilities_test.go
index 8d2fdaa9..e2a62998 100644
--- a/internal/serverapp/capabilities_test.go
+++ b/internal/serverapp/capabilities_test.go
@@ -57,8 +57,8 @@ func TestAdmissionCapabilitiesMatchHealthAndRoutes(t *testing.T) {
options.Queue = store.Queue()
}
if test.collect {
- options.Collections = serverapp.NewCollections(service, func(context.Context) (serverapp.Collection, error) {
- return serverapp.Collection{}, errors.New("unused")
+ options.Collections = serverapp.NewCollections(service, func(context.Context, func(ingestion.Request) error) ([]serverapp.CollectionFailure, error) {
+ return nil, errors.New("unused")
}, "")
}
if got := options.Capabilities(); got != test.want {
@@ -179,7 +179,7 @@ func TestCollectionIngestsEveryCheckoutAndMovesMCPDefault(t *testing.T) {
roots := []string{root}
var failure error
var failed []serverapp.CollectionFailure
- collect := func(ctx context.Context) (serverapp.Collection, error) {
+ collectBatch := func(ctx context.Context) (serverapp.Collection, error) {
mu.Lock()
defer mu.Unlock()
if failure != nil {
@@ -196,7 +196,7 @@ func TestCollectionIngestsEveryCheckoutAndMovesMCPDefault(t *testing.T) {
}
return collection, nil
}
- initial, err := collect(t.Context())
+ initial, err := collectBatch(t.Context())
if err != nil {
t.Fatal(err)
}
@@ -219,6 +219,18 @@ func TestCollectionIngestsEveryCheckoutAndMovesMCPDefault(t *testing.T) {
if !ok {
t.Fatal("initial collection published nothing")
}
+ collect := func(ctx context.Context, publish func(ingestion.Request) error) ([]serverapp.CollectionFailure, error) {
+ collection, err := collectBatch(ctx)
+ if err != nil {
+ return nil, err
+ }
+ for _, input := range collection.Requests {
+ if err := publish(input); err != nil {
+ return nil, err
+ }
+ }
+ return collection.Failures, nil
+ }
server := httptest.NewServer(serverapp.Handler(t.Context(), service, store, fstest.MapFS{}, first.Context.ID, serverapp.Options{Collections: serverapp.NewCollections(service, collect, first.Context.ID)}))
defer server.Close()
d := deployment{server: server}
diff --git a/internal/serverapp/collection.go b/internal/serverapp/collection.go
index ca1ad13f..ac11e65a 100644
--- a/internal/serverapp/collection.go
+++ b/internal/serverapp/collection.go
@@ -45,6 +45,11 @@ type CollectionFailure struct {
Detail string `json:"detail"`
}
+// Collect publishes completed checkout requests serially before returning the
+// remaining checkout failures. It joins its workers before returning, including
+// after cancellation or a publication error.
+type Collect func(context.Context, func(ingestion.Request) error) ([]CollectionFailure, error)
+
// Published is the outcome of ingesting one collection.
type Published struct {
Submissions []contextservice.Submission
@@ -84,21 +89,28 @@ func (published Published) Newest() (contextservice.Submission, bool) {
func Publish(ctx context.Context, service *contextservice.Service, collection Collection) (Published, error) {
published := Published{Failures: slices.Clone(collection.Failures)}
for _, input := range collection.Requests {
- service.Select(input)
- submitted, err := service.Ingest(ctx, input)
- var rejected *review.RequestError
- if errors.As(err, &rejected) {
- published.Failures = append(published.Failures, CollectionFailure{Path: input.Metadata.Root, Detail: rejected.Detail})
- continue
- }
- if err != nil {
+ if err := published.ingest(ctx, service, input); err != nil {
return published, err
}
- published.Submissions = append(published.Submissions, submitted)
}
return published, nil
}
+func (published *Published) ingest(ctx context.Context, service *contextservice.Service, input ingestion.Request) error {
+ service.Select(input)
+ submitted, err := service.Ingest(ctx, input)
+ var rejected *review.RequestError
+ if errors.As(err, &rejected) {
+ published.Failures = append(published.Failures, CollectionFailure{Path: input.Metadata.Root, Detail: rejected.Detail})
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ published.Submissions = append(published.Submissions, submitted)
+ return nil
+}
+
// FailureSummary joins failures for a single diagnostic.
func FailureSummary(failures []CollectionFailure) string {
details := make([]string, 0, len(failures))
@@ -117,7 +129,7 @@ type Collections struct {
pending *collectionResult
gate chan struct{}
service *contextservice.Service
- collect func(context.Context) (Collection, error)
+ collect Collect
current *defaultContext
}
@@ -127,7 +139,7 @@ type collectionResult struct {
err error
}
-func NewCollections(service *contextservice.Service, collect func(context.Context) (Collection, error), defaultID string) *Collections {
+func NewCollections(service *contextservice.Service, collect Collect, defaultID string) *Collections {
return &Collections{service: service, collect: collect, current: &defaultContext{id: defaultID}, gate: make(chan struct{}, 1)}
}
@@ -158,14 +170,27 @@ func (handler *Collections) run(ctx context.Context) (Published, error) {
if err := ctx.Err(); err != nil {
return Published{}, err
}
- collected, err := handler.collect(ctx)
- if err != nil {
- return Published{}, &review.RequestError{Status: http.StatusUnprocessableEntity, Detail: "Unable to collect checkouts: " + err.Error()}
+ published := Published{}
+ var publishError error
+ failures, err := handler.collect(ctx, func(input ingestion.Request) error {
+ if publishError = published.ingest(ctx, handler.service, input); publishError != nil {
+ return publishError
+ }
+ if newest, ok := published.Newest(); ok {
+ handler.current.set(newest.Context.ID)
+ }
+ return nil
+ })
+ published.Failures = append(published.Failures, failures...)
+ if publishError != nil {
+ return published, publishError
}
- published, err := Publish(ctx, handler.service, collected)
- if err != nil {
+ if err := ctx.Err(); err != nil {
return published, err
}
+ if err != nil {
+ return published, &review.RequestError{Status: http.StatusUnprocessableEntity, Detail: "Unable to collect checkouts: " + err.Error()}
+ }
newest, ok := published.Newest()
if !ok {
return published, &review.RequestError{Status: http.StatusUnprocessableEntity, Detail: "Unable to collect checkouts: " + FailureSummary(published.Failures)}
diff --git a/internal/serverapp/collection_test.go b/internal/serverapp/collection_test.go
index aa4047f7..c3c88c2b 100644
--- a/internal/serverapp/collection_test.go
+++ b/internal/serverapp/collection_test.go
@@ -71,9 +71,16 @@ func TestOlderCollectionRemainsSuccessfulWithoutRegressingStreamHead(t *testing.
if !ok || fallback.Context.ID != published.Submissions[0].Context.ID {
t.Fatalf("successful older collection lost its default: %+v, %v", fallback, ok)
}
- handler := NewCollections(service, func(ctx context.Context) (Collection, error) {
+ var handler *Collections
+ handler = NewCollections(service, func(_ context.Context, publish func(ingestion.Request) error) ([]CollectionFailure, error) {
request.SubmissionID = "earlier-clock-again"
- return Collection{Requests: []ingestion.Request{request}}, nil
+ if err := publish(request); err != nil {
+ return nil, err
+ }
+ if handler.current.get() != fallback.Context.ID {
+ t.Error("MCP default waited for collection completion")
+ }
+ return nil, nil
}, later.Context.ID)
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/api/v2/collections", nil))
diff --git a/internal/serverapp/startup_collection_test.go b/internal/serverapp/startup_collection_test.go
index 45ca89c3..8ccbe4df 100644
--- a/internal/serverapp/startup_collection_test.go
+++ b/internal/serverapp/startup_collection_test.go
@@ -6,13 +6,19 @@ import (
"errors"
"net/http"
"net/http/httptest"
+ "os"
+ "os/exec"
+ "path/filepath"
"strings"
+ "sync"
"sync/atomic"
"testing"
"testing/fstest"
"time"
+ "github.com/flexdinesh/diffx/internal/collector"
"github.com/flexdinesh/diffx/internal/contextservice"
+ "github.com/flexdinesh/diffx/internal/diffsource"
"github.com/flexdinesh/diffx/internal/ingestion"
"github.com/flexdinesh/diffx/internal/reviewstore"
"github.com/flexdinesh/diffx/internal/serverapp"
@@ -94,14 +100,14 @@ func TestInitialCollectionJoinsFirstRequestAndLaterRequestsCollectAgain(t *testi
service, _ := collectionService(t)
started, release := make(chan struct{}), make(chan struct{})
var calls atomic.Int32
- collections := serverapp.NewCollections(service, func(ctx context.Context) (serverapp.Collection, error) {
+ collections := serverapp.NewCollections(service, func(ctx context.Context, publish func(ingestion.Request) error) ([]serverapp.CollectionFailure, error) {
call := calls.Add(1)
if call == 1 {
close(started)
select {
case <-release:
case <-ctx.Done():
- return serverapp.Collection{}, ctx.Err()
+ return nil, ctx.Err()
}
}
fresh := input
@@ -109,7 +115,7 @@ func TestInitialCollectionJoinsFirstRequestAndLaterRequestsCollectAgain(t *testi
fresh.SubmissionID = "explicit-sync"
fresh.Metadata.CollectedAt++
}
- return serverapp.Collection{Requests: []ingestion.Request{fresh}}, nil
+ return nil, publish(fresh)
}, "")
completed := runInitialCollection(t, collections)
select {
@@ -161,11 +167,11 @@ func TestInitialCollectionFailureIsReportedThenRetryCollectsAgain(t *testing.T)
input := collectGit(t)
service, _ := collectionService(t)
var calls atomic.Int32
- collections := serverapp.NewCollections(service, func(context.Context) (serverapp.Collection, error) {
+ collections := serverapp.NewCollections(service, func(_ context.Context, publish func(ingestion.Request) error) ([]serverapp.CollectionFailure, error) {
if calls.Add(1) == 1 {
- return serverapp.Collection{}, errors.New("checkout unavailable")
+ return nil, errors.New("checkout unavailable")
}
- return serverapp.Collection{Requests: []ingestion.Request{input}}, nil
+ return nil, publish(input)
}, "")
completed := runInitialCollection(t, collections)
if err := awaitCollection(t, completed); err == nil || !strings.Contains(err.Error(), "checkout unavailable") {
@@ -191,13 +197,13 @@ func TestExplicitSyncCollectsFreshBeforeAnyBrowserInitialRequest(t *testing.T) {
input := collectGit(t)
service, _ := collectionService(t)
var calls atomic.Int32
- collections := serverapp.NewCollections(service, func(context.Context) (serverapp.Collection, error) {
+ collections := serverapp.NewCollections(service, func(_ context.Context, publish func(ingestion.Request) error) ([]serverapp.CollectionFailure, error) {
fresh := input
if calls.Add(1) > 1 {
fresh.SubmissionID = "explicit-sync"
fresh.Metadata.CollectedAt++
}
- return serverapp.Collection{Requests: []ingestion.Request{fresh}}, nil
+ return nil, publish(fresh)
}, "")
completed := runInitialCollection(t, collections)
if err := awaitCollection(t, completed); err != nil {
@@ -216,16 +222,16 @@ func TestBrowserCancellationDoesNotCancelInitialCollection(t *testing.T) {
service, _ := collectionService(t)
started, release := make(chan struct{}), make(chan struct{})
var calls atomic.Int32
- collections := serverapp.NewCollections(service, func(ctx context.Context) (serverapp.Collection, error) {
+ collections := serverapp.NewCollections(service, func(ctx context.Context, publish func(ingestion.Request) error) ([]serverapp.CollectionFailure, error) {
if calls.Add(1) > 1 {
- return serverapp.Collection{Requests: []ingestion.Request{input}}, nil
+ return nil, publish(input)
}
close(started)
select {
case <-release:
- return serverapp.Collection{Requests: []ingestion.Request{input}}, nil
+ return nil, publish(input)
case <-ctx.Done():
- return serverapp.Collection{}, ctx.Err()
+ return nil, ctx.Err()
}
}, "")
completed := runInitialCollection(t, collections)
@@ -261,8 +267,8 @@ func TestCollectionScopeStartsEmptyAndPublishesOnlySelectedCheckouts(t *testing.
if err != nil {
t.Fatal(err)
}
- collections := serverapp.NewCollections(service, func(context.Context) (serverapp.Collection, error) {
- return serverapp.Collection{Requests: []ingestion.Request{input}}, nil
+ collections := serverapp.NewCollections(service, func(_ context.Context, publish func(ingestion.Request) error) ([]serverapp.CollectionFailure, error) {
+ return nil, publish(input)
}, "")
assets := fstest.MapFS{"index.html": {Data: []byte("review UI")}}
handler := serverapp.Handler(t.Context(), service, store, assets, "", serverapp.Options{Collections: collections})
@@ -295,3 +301,97 @@ func TestCollectionScopeStartsEmptyAndPublishesOnlySelectedCheckouts(t *testing.
t.Fatalf("unrelated review resolved: %d %s", response.Code, response.Body.String())
}
}
+
+func TestInitialCollectionMakesReadyCheckoutReadableWhileAnotherIsCollecting(t *testing.T) {
+ repository := func() string {
+ root := t.TempDir()
+ if err := os.WriteFile(filepath.Join(root, "file.txt"), []byte("before\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ for _, args := range [][]string{{"init", "-q"}, {"add", "."}, {"-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "initial"}} {
+ if raw, err := exec.Command("git", append([]string{"-C", root}, args...)...).CombinedOutput(); err != nil {
+ t.Fatalf("git %v: %v %s", args, err, raw)
+ }
+ }
+ if err := os.WriteFile(filepath.Join(root, "file.txt"), []byte("after\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ return root
+ }
+ first, second := repository(), repository()
+ service, store := collectionService(t)
+ blocked, release := make(chan struct{}), make(chan struct{})
+ var releaseOnce sync.Once
+ t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) })
+ var observed int
+ collections := serverapp.NewCollections(service, func(ctx context.Context, publish func(ingestion.Request) error) ([]serverapp.CollectionFailure, error) {
+ options := collector.Options{SourceID: "streaming-startup", OnSource: func(ingestion.Metadata, diffsource.Comparison) {
+ observed++
+ if observed == 2 {
+ close(blocked)
+ select {
+ case <-release:
+ case <-ctx.Done():
+ }
+ }
+ }}
+ batch, err := collector.CollectWorkspaces(ctx, []string{first, second}, 8, options, publish)
+ failures := make([]serverapp.CollectionFailure, 0, len(batch.Failures))
+ for _, failure := range batch.Failures {
+ failures = append(failures, serverapp.CollectionFailure{Path: failure.Path, Detail: failure.Detail})
+ }
+ return failures, err
+ }, "")
+ server := httptest.NewServer(serverapp.Handler(t.Context(), service, store, fstest.MapFS{}, "", serverapp.Options{Collections: collections}))
+ defer server.Close()
+ deployment := deployment{server: server}
+ events, unsubscribe := service.Subscribe()
+ defer unsubscribe()
+ completed := runInitialCollection(t, collections)
+ select {
+ case <-blocked:
+ case <-time.After(5 * time.Second):
+ t.Fatal("second checkout did not block")
+ }
+ select {
+ case <-events:
+ case <-time.After(5 * time.Second):
+ t.Fatal("ready checkout did not publish")
+ }
+ page := decode[contextservice.Page](t, request(t, deployment, http.MethodGet, "/api/v2/contexts", nil))
+ if len(page.Contexts) != 1 {
+ t.Fatalf("partial catalog: %#v", page.Contexts)
+ }
+ id := page.Contexts[0].ID
+ for _, path := range []string{"/api/v2/contexts/" + id, "/api/v2/contexts/" + id + "/diffs/current?scope=all"} {
+ response := request(t, deployment, http.MethodGet, path, nil)
+ response.Body.Close()
+ if response.StatusCode != http.StatusOK {
+ t.Fatalf("ready checkout %s: status %d", path, response.StatusCode)
+ }
+ }
+ select {
+ case err := <-completed:
+ t.Fatalf("collection completed before held checkout: %v", err)
+ default:
+ }
+ releaseOnce.Do(func() { close(release) })
+ if err := awaitCollection(t, completed); err != nil {
+ t.Fatal(err)
+ }
+ response := httptest.NewRecorder()
+ collections.ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/api/v2/collections?initial=true", nil))
+ if response.Code != http.StatusOK {
+ t.Fatalf("complete response: %d %s", response.Code, response.Body.String())
+ }
+ var result struct {
+ Contexts []contextservice.Context `json:"contexts"`
+ Failures []serverapp.CollectionFailure `json:"failures"`
+ }
+ if err := json.Unmarshal(response.Body.Bytes(), &result); err != nil {
+ t.Fatal(err)
+ }
+ if len(result.Contexts) != 2 || len(result.Failures) != 0 {
+ t.Fatalf("aggregate result: %#v", result)
+ }
+}
diff --git a/packages/api/src/client.ts b/packages/api/src/client.ts
index 13daaba7..de5256ef 100644
--- a/packages/api/src/client.ts
+++ b/packages/api/src/client.ts
@@ -20,6 +20,7 @@ export type ApiFilePatch = components["schemas"]["FilePatch"];
export type ApiFileContents = components["schemas"]["FileContents"];
export type ApiReviewComment = components["schemas"]["ReviewComment"];
export type ApiReviewMark = components["schemas"]["ReviewMark"];
+export type ApiCollectionResult = components["schemas"]["CollectionResult"];
export type ApiCollectionFailure = components["schemas"]["CollectionFailure"];
export function errorDetail(error: unknown, fallback: string) {