[Hub] Dedupe file entries by xet hash within a shard (#2134)

## Summary
- Track xet file hashes already written into the current shard's
file-info section and skip duplicates so the shard's file list behaves
like a set rather than a list.
- The dedup set is scoped to the entire `uploadShards` call so it
persists across shard flushes — a duplicate in a later shard is also
skipped.
- File events are still yielded to callers for every input path, so
consumers (e.g. `commit.ts`) keep getting per-path progress and `path ->
xet hash` mapping for duplicates.

Today the only dedup that happens is sha256-based and lives in
`createXorbs` — if a file has no sha256 (or only one of the duplicates
does), both entries currently get written into the shard's file list.
This change handles that case.

## Test plan
- [x] `npx vitest run src/utils/uploadShards.spec.ts` — added a case
that feeds three paths through `uploadShards` with the mocked
`createXorbs` yielding the same xet hash, and asserts: each path still
gets a `file` event, but only one file entry ends up in the resulting
shard.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes shard serialization behavior by skipping duplicate file
entries based on Xet hash, which could impact upload correctness and
downstream consumers if the dedupe scope/ordering assumptions are wrong.
> 
> **Overview**
> Makes `uploadShards` treat the shard file-info section like a set by
tracking seen Xet file hashes and **skipping writing duplicate file
entries** while still yielding `file` events for every input path.
> 
> Adds a new unit test that feeds multiple paths producing the same Xet
hash and asserts only one file entry is stored in the uploaded shard,
while all paths still receive `file` events.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
24884a1. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Co-authored-by: Eliott C. <coyotte508@gmail.com>

Author: assafvayner

packages/hub/src/utils/uploadShards.spec.ts

@@ -91,6 +91,17 @@ function toSource(sha256?: string): AsyncGenerator<{ content: Blob; path: string
 	})();
 }
 
+function toMultiSource(paths: string[]): AsyncGenerator<{ content: Blob; path: string; sha256?: string }> {
+	return (async function* () {
+		for (const path of paths) {
+			yield {
+				content: new Blob(["content"]),
+				path,
+			};
+		}
+	})();
+}
+
 describe("uploadShards", () => {
 	it("omits metadata flag and metadata section when sha256 is missing", async () => {
 		const uploadedShards: Uint8Array[] = [];
@@ -167,4 +178,53 @@ describe("uploadShards", () => {
 			fileEntryLength: FILE_ENTRY_BASE_SIZE + REPRESENTATION_ENTRY_SIZE + VERIFICATION_ENTRY_SIZE + METADATA_ENTRY_SIZE,
 		});
 	});
+
+	it("dedupes file entries with the same xet hash within a shard", async () => {
+		const uploadedShards: Uint8Array[] = [];
+		const fetchMock: typeof fetch = vi.fn(async (input, init) => {
+			const url = String(input);
+
+			if (url.endsWith("/v1/shards")) {
+				if (!(init?.body instanceof Uint8Array)) {
+					throw new Error("Expected Uint8Array shard body");
+				}
+				uploadedShards.push(new Uint8Array(init.body));
+			}
+
+			return new Response(null, { status: 200 });
+		});
+
+		const fileEvents: Array<{ path: string; xetHash: string }> = [];
+		for await (const event of uploadShards(toMultiSource(["a.bin", "b.bin", "c.bin"]), {
+			accessToken: "test-token",
+			hubUrl: "https://hub.local",
+			fetch: fetchMock,
+			repo: { type: "model", name: "user/repo" },
+			rev: "main",
+			xetParams: {
+				casUrl: "https://cas.local",
+				accessToken: "cas-token",
+				expiresAt: new Date(Date.now() + 600_000),
+				refreshWriteTokenUrl: "https://hub.local/xet-write-token",
+			},
+		})) {
+			if (event.event === "file") {
+				fileEvents.push({ path: event.path, xetHash: event.xetHash });
+			}
+		}
+
+		// Each path still gets its file event yielded so callers can map path -> hash.
+		expect(fileEvents).toEqual([
+			{ path: "a.bin", xetHash: "3".repeat(64) },
+			{ path: "b.bin", xetHash: "3".repeat(64) },
+			{ path: "c.bin", xetHash: "3".repeat(64) },
+		]);
+
+		// But only one file entry is written into the shard.
+		expect(uploadedShards).toHaveLength(1);
+		expect(readFileEntryInfo(uploadedShards[0])).toEqual({
+			flags: MDB_FILE_FLAG_WITH_VERIFICATION,
+			fileEntryLength: FILE_ENTRY_BASE_SIZE + REPRESENTATION_ENTRY_SIZE + VERIFICATION_ENTRY_SIZE,
+		});
+	});
 });

packages/hub/src/utils/uploadShards.ts

@@ -87,6 +87,7 @@ export async function* uploadShards(
 	| { event: "fileProgress"; path: string; progress: number }
 > {
 	const xorbHashes: Array<string> = [];
+	const seenFileXetHashes = new Set<string>();
 
 	const fileInfoSection = new Uint8Array(Math.floor(SHARD_MAX_SIZE - SHARD_HEADER_SIZE - SHARD_FOOTER_SIZE) * 0.25);
 	const xorbInfoSection = new Uint8Array(Math.floor(SHARD_MAX_SIZE - SHARD_HEADER_SIZE - SHARD_FOOTER_SIZE) * 0.75);
@@ -172,6 +173,11 @@ export async function* uploadShards(
 					dedupRatio: output.dedupRatio,
 				}; // Maybe wait until shard is uploaded before yielding.
 
+				if (seenFileXetHashes.has(output.hash)) {
+					break;
+				}
+				seenFileXetHashes.add(output.hash);
+
 				// Calculate space needed for this file entry
 				const fileHeaderSize = HASH_LENGTH + 4 + 4 + 8; // hash + flags + rep length + reserved
 				const representationSize = output.representation.length * (HASH_LENGTH + 4 + 4 + 4 + 4); // per rep: xorb hash + flags + length + offset + endOffset