test: verify ReplicationHandler abort lifecycle

test/integration/solr-replication-lifecycle.test.ts

@@ -1,4 +1,5 @@
 import assert from "node:assert/strict";
+import { randomBytes } from "node:crypto";
 import { SolrClient } from "../../src/index.js";
 import { integrationTest } from "./fixtures.js";
 
@@ -77,3 +78,91 @@ integrationTest("ReplicationHandler fetches a committed leader index into an iso
   const followerGeneration = followerVersion.generation ?? Number(followerVersion.indexversion);
   assert.ok(Number.isFinite(followerGeneration) && followerGeneration > 0, "follower should publish its fetched index");
 });
+
+integrationTest("ReplicationHandler aborts an in-progress fetch and reports its state", async ({
+  createCore,
+  createId,
+}) => {
+  const leaderCore = await createCore("replication_abort_leader");
+  const followerCore = await createCore("replication_abort_follower");
+  const baseUrl = process.env.SOLR_BASE_URL!;
+  const username = process.env.SOLR_USERNAME;
+  const password = process.env.SOLR_PASSWORD;
+  const makeClient = (core: string) => new SolrClient({
+    baseUrl,
+    core,
+    timeoutMs: 120_000,
+    ...(username && password ? { auth: { type: "basic" as const, username, password } } : {})
+  });
+  const leader = makeClient(leaderCore);
+  const follower = makeClient(followerCore);
+  const leaderUrl = `${baseUrl.replace(/\/$/, "")}/${leaderCore}`;
+  const documentId = createId("replication_abort");
+  const storedField = createId("replication_payload");
+  const payload = randomBytes(12 * 1024 * 1024).toString("base64");
+
+  await leader.schema({ "add-field": { name: storedField, type: "string", stored: true, indexed: false, docValues: false } });
+  await leader.add([{ id: documentId, [storedField]: payload }]);
+  await leader.commit({ waitSearcher: true });
+
+  await leader.config({
+    "update-requesthandler": {
+      name: "/replication",
+      class: "solr.ReplicationHandler",
+      customOptions: {
+        leader: { replicateAfter: ["commit"] },
+        // Throttle leader file streaming so the follower remains in a real in-progress fetch.
+        defaults: { maxWriteMBPerSec: 0.05 }
+      }
+    }
+  });
+  await follower.config({
+    "update-requesthandler": {
+      name: "/replication",
+      class: "solr.ReplicationHandler",
+      customOptions: { follower: { leaderUrl, ...(username && password ? { httpBasicAuthUser: username, httpBasicAuthPassword: password } : {}) } }
+    }
+  });
+
+  const fetchResponse = await follower.replication("fetchindex", {
+    leaderUrl,
+    ...(username && password ? { customParams: { httpBasicAuthUser: username, httpBasicAuthPassword: password } } : {})
+  });
+  assert.equal(fetchResponse.status, "OK", "fetchindex should start asynchronously");
+
+  const isReplicating = async (): Promise<boolean> => {
+    const response = await follower.replication("details");
+    const details = response.details as { follower?: { isReplicating?: string | boolean } } | undefined;
+    return String(details?.follower?.isReplicating).toLowerCase() === "true";
+  };
+
+  let observedActiveFetch = false;
+  for (let attempt = 0; attempt < 200; attempt += 1) {
+    if (await isReplicating()) {
+      observedActiveFetch = true;
+      break;
+    }
+    await new Promise(resolve => setTimeout(resolve, 25));
+  }
+  assert.ok(observedActiveFetch, "details should expose isReplicating=true before the large fetch completes");
+
+  const abortResponse = await follower.replication("abortfetch");
+  assert.equal(abortResponse.status, "OK", "abortfetch should acknowledge the cancellation request");
+
+  let fetchStopped = false;
+  for (let attempt = 0; attempt < 200; attempt += 1) {
+    if (!(await isReplicating())) {
+      fetchStopped = true;
+      break;
+    }
+    await new Promise(resolve => setTimeout(resolve, 25));
+  }
+  assert.ok(fetchStopped, "details should report that the fetch stopped after abortfetch");
+
+  const result = await follower.query<{ response?: { numFound?: number } }>({
+    query: `id:"${documentId}"`,
+    fields: ["id"],
+    limit: 1
+  });
+  assert.equal(result.response?.numFound ?? 0, 0, "an aborted fetch must not install the leader document");
+});