PROD-3025, PROD-3076: push linked content in dependency order, two-pass for reference cycles

ci/permutation/fixtures.mjs

@@ -104,6 +104,52 @@ export async function seedReferenceCycles(c, guid, locale) {
   };
 }
 
+// Same rule as the CLI's Auth.determineBaseUrl / determineFetchUrl: the GUID suffix picks the region.
+const REGIONS = [
+  ["us2", "mgmt-usa2", "api-usa2"],
+  ["d", "mgmt-dev", "api-dev"],
+  ["u", "mgmt", "api"],
+  ["c", "mgmt-ca", "api-ca"],
+  ["e", "mgmt-eu", "api-eu"],
+  ["a", "mgmt-aus", "api-aus"],
+];
+function hosts(guid) {
+  const [, mgmt, api] = REGIONS.find(([suffix]) => guid.endsWith(suffix)) ?? ["", "mgmt", "api"];
+  return { mgmt: `https://${mgmt}.aglty.io`, api: `https://${api}.aglty.io` };
+}
+
+// Content published through the Management API reaches the Fetch API's sync feed a minute or two
+// later, and the CLI pulls from that feed. Wait until it lists every given item, or the timeout.
+// Returns the seconds waited; throws on timeout.
+export async function waitForFetchSync(token, guid, locale, ids, { timeoutMs = 10 * 60 * 1000 } = {}) {
+  const { mgmt, api } = hosts(guid);
+  const keyRes = await fetch(`${mgmt}/oauth/GetPreviewKey?guid=${guid}`, {
+    headers: { Authorization: `Bearer ${token}`, "Cache-Control": "no-cache" },
+  });
+  if (!keyRes.ok) throw new Error(`GetPreviewKey: HTTP ${keyRes.status}`);
+  const apiKey = (await keyRes.text()).replace(/^"|"$/g, "");
+
+  const started = Date.now();
+  const missing = new Set(ids);
+  while (Date.now() - started < timeoutMs) {
+    // Read the whole feed from the start, a page at a time, as a full pull does
+    let syncToken = 0;
+    for (let page = 0; page < 100 && missing.size > 0; page++) {
+      const res = await fetch(`${api}/${guid}/preview/${locale}/sync/items?syncToken=${syncToken}&pageSize=500`, {
+        headers: { APIKey: apiKey, "Cache-Control": "no-cache" },
+      });
+      if (!res.ok) throw new Error(`Fetch API sync/items: HTTP ${res.status}`);
+      const body = await res.json();
+      for (const item of body.items ?? []) missing.delete(item.contentID);
+      if (!body.items?.length || !body.syncToken || body.syncToken === syncToken) break;
+      syncToken = body.syncToken;
+    }
+    if (missing.size === 0) return Math.round((Date.now() - started) / 1000);
+    await new Promise((r) => setTimeout(r, 10_000));
+  }
+  throw new Error(`items ${[...missing].join(", ")} not in the Fetch API sync feed after ${timeoutMs / 1000}s`);
+}
+
 // After a sync: each target copy's Next (and NextID) must hold the TARGET ID of its partner.
 export async function checkReferenceCycles(c, guid, locale, fixture, toTarget) {
   const problems = [];

ci/permutation/run.mjs

@@ -27,7 +27,7 @@ import { spawnSync } from "node:child_process";
 import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
 import { join, resolve } from "node:path";
 import { CHANGES, QUICK_ROUNDS, client, conflict, drift, readField, rng } from "./changes.mjs";
-import { checkReferenceCycles, seedReferenceCycles } from "./fixtures.mjs";
+import { checkReferenceCycles, seedReferenceCycles, waitForFetchSync } from "./fixtures.mjs";
 
 const { SOURCE_GUID: A, TARGET_GUID: B, AGILITY_TOKEN, CLI } = process.env;
 const TIER = process.env.TIER || "quick";
@@ -130,12 +130,19 @@ function syncAndCheck(round, direction) {
 let cycles = null;
 try {
   cycles = await seedReferenceCycles(api, A, LOCALE);
+  // The first sync pulls from the Fetch API, which lists new items a minute or two after publish
+  const waited = await waitForFetchSync(
+    AGILITY_TOKEN,
+    A,
+    LOCALE,
+    cycles.links.map(([x]) => x)
+  );
   record(
     0,
     "A→B",
     "apply:fixture-cycles",
     true,
-    `A ↔ B and a self-reference (${cycles.links.map(([x]) => `#${x}`).join(", ")})`
+    `A ↔ B and a self-reference (${cycles.links.map(([x]) => `#${x}`).join(", ")}), in the Fetch API after ${waited}s`
   );
 } catch (e) {
   record(0, "A→B", "apply:fixture-cycles", false, String(e?.message ?? e).slice(0, 200));

src/lib/mappers/content-item-mapper.ts

@@ -159,6 +159,18 @@ export class ContentItemMapper {
     this.saveMapping();
   }
 
+  /**
+   * PROD-3076: make the next sync treat this item's source as changed, so it is pushed again. Used
+   * when an item was created without its references (pass 1 of a reference cycle) and the update
+   * that fills them in failed.
+   */
+  markSourceChanged(sourceContentID: number): void {
+    const mapping = this.getContentItemMappingByContentID(sourceContentID, "source");
+    if (!mapping) return;
+    mapping.sourceVersionID = 0;
+    this.saveMapping();
+  }
+
   updateMapping(
     sourceContentItem: mgmtApi.ContentItem,
     targetContentItem: mgmtApi.ContentItem,

src/lib/pushers/content-pusher/content-batch-processor.ts

@@ -495,7 +495,13 @@ export class ContentBatchProcessor {
           const isTargetContentItemDeleted = existingMapping && existingTargetContentItem == null;
           let existingContentID = -1;
 
-          if (isTargetContentItemDeleted && state.overwrite) {
+          // PROD-3076: pass 2 of a reference cycle updates an item pass 1 created moments ago; it is
+          // not in the pulled target data yet, so without this it would look deleted and be recreated.
+          const createdThisRun = !!this.config.trustSameRunMappings && !!existingMapping && !existingTargetContentItem;
+
+          if (createdThisRun) {
+            existingContentID = existingMapping.targetContentID;
+          } else if (isTargetContentItemDeleted && state.overwrite) {
             existingContentID = existingMapping ? existingMapping?.targetContentID : -1;
           } else {
             existingContentID = existingTargetContentItem ? existingTargetContentItem.contentID : -1;
@@ -585,7 +591,9 @@ export class ContentBatchProcessor {
             // Only an update of a target item known to exist (or a locale version of one found in
             // another locale) must come back with the same ID.
             expectedId:
-              existingTargetContentItem || (!existingMapping && existingContentID > 0) ? existingContentID : undefined,
+              existingTargetContentItem || createdThisRun || (!existingMapping && existingContentID > 0)
+                ? existingContentID
+                : undefined,
           });
         } catch (error: any) {
           const prepareError = error.message || "payload preparation failed";

src/lib/pushers/content-pusher/content-pusher.ts

@@ -10,6 +10,8 @@ import { ContentItem, Model } from "@agility/management-sdk";
 import { ContainerMapper } from "lib/mappers/container-mapper";
 import { getApiClient } from "core/state";
 import { reportForeignCdnReferences } from "./util/report-foreign-cdn-references";
+import { BatchProcessingResult } from "./util/types";
+import { stripContentReferences } from "./util/strip-content-references";
 
 /**
  * Push content to the target instance
@@ -33,13 +35,15 @@ export async function pushContent(sourceData: ContentItem[], targetData: Content
     return { status: "success" as const, successful: 0, failed: 0, skipped: 0, publishableIds: [], failureDetails: [] };
   }
 
-  // Deterministically classify content items based on list references (fulllist=true)
-  const { normalContentItems, linkedContentItems, skippedItems } = getContentItemTypes(contentItems, {
-    containerMapper,
-    modelMapper,
-    referenceMapper,
-    logger: logger as any,
-  });
+  // Deterministically classify content items by references: linked (referenced, split into
+  // dependency levels), normal (unreferenced) and skipped (missing mappings)
+  const { normalContentItems, linkedContentItems, linkedContentLevels, linkedCycleReferences, skippedItems } =
+    getContentItemTypes(contentItems, {
+      containerMapper,
+      modelMapper,
+      referenceMapper,
+      logger: logger as any,
+    });
 
   // PROD-2773: warn about asset URLs into a CDN container owned by neither instance. Reporting
   // only — runs before change detection (and in --preflight) and never alters an item. Items
@@ -99,6 +103,37 @@ export async function pushContent(sourceData: ContentItem[], targetData: Content
     }
   };
 
+  // Fold one processBatches result into the totals; register each failure so the next pass's
+  // parents and the page pusher can say why their reference is unresolved. Items in `heldBack`
+  // were saved as a pass-1 stub (PROD-3076): pass 2 counts them, and only pass 2's version is
+  // published, so their successes here are left out.
+  const recordBatchResult = (result: BatchProcessingResult, heldBack: Set<number> = new Set()): void => {
+    const heldBackTargetIds = new Set(
+      (result.successfulItems || [])
+        .filter((item) => heldBack.has(item.originalContent?.contentID))
+        .map((item) => item.newContentId)
+    );
+    totalSuccessful += result.successCount - heldBackTargetIds.size;
+    totalFailed += result.failureCount;
+    totalSkipped += result.skippedCount;
+    allPublishableIds.push(...result.publishableIds.filter((id) => !heldBackTargetIds.has(id)));
+    for (const item of result.failedItems || []) {
+      const name = item.originalContent?.properties?.referenceName || "Unknown";
+      const contentID = item.originalContent?.contentID;
+      allFailureDetails.push({
+        name,
+        error: item.error,
+        type: "content",
+        contentID,
+        guid: sourceGuidStr,
+        locale,
+      });
+      if (contentID) {
+        registerFailedContent(contentID, name, item.error, locale);
+      }
+    }
+  };
+
   try {
     // Import getApiClient for both batch configurations
 
@@ -107,7 +142,8 @@ export async function pushContent(sourceData: ContentItem[], targetData: Content
       totalSkipped += skippedItems.length;
     }
 
-    // Process linked content items second (with dependencies)
+    // Process linked content items first, one dependency level at a time (leaves first), so
+    // the mappings each level references were written by an earlier level (PROD-3025).
     if (linkedContentItems.length > 0) {
       const linkedBatchConfig = {
         apiClient: getApiClient(),
@@ -139,37 +175,31 @@ export async function pushContent(sourceData: ContentItem[], targetData: Content
         totalSuccessful += filteredLinkedContentItems.itemsToProcess.length;
         totalSkipped += filteredLinkedContentItems.skippedCount;
       } else {
-      const linkedBatchProcessor = new ContentBatchProcessor(linkedBatchConfig);
-      const linkedResult = await linkedBatchProcessor.processBatches(
-        filteredLinkedContentItems.itemsToProcess.reverse(),
-        logger,
-        "Linked Content"
-      );
+        totalSkipped += filteredLinkedContentItems.skippedCount;
+        const levels = splitIntoLevels(filteredLinkedContentItems.itemsToProcess, linkedContentLevels);
+        const linkedBatchProcessor = new ContentBatchProcessor(linkedBatchConfig);
+        const cycleBatchProcessor = new ContentBatchProcessor({ ...linkedBatchConfig, trustSameRunMappings: true });
+        for (let i = 0; i < levels.length; i++) {
+          const label = levels.length > 1 ? `Linked Content (level ${i + 1}/${levels.length})` : "Linked Content";
+
+          // PROD-3076: items in a reference cycle go out first with their references to not-yet-
+          // created cycle members blanked, then again in full once every member has a target ID.
+          const { toPush, stubbed } = stubCycleReferences(levels[i], linkedCycleReferences, referenceMapper, modelMapper);
+          const result = await linkedBatchProcessor.processBatches(toPush, logger, label);
+          recordBatchResult(result, new Set(stubbed.map((item) => item.contentID)));
 
-      totalSuccessful += linkedResult.successCount;
-      totalFailed += linkedResult.failureCount;
-      totalSkipped += filteredLinkedContentItems.skippedCount;
-      totalSkipped += linkedResult.skippedCount;
-      allPublishableIds.push(...linkedResult.publishableIds);
-      // Collect failure details for error summary and register in failed content registry
-      if (linkedResult.failedItems && linkedResult.failedItems.length > 0) {
-        linkedResult.failedItems.forEach((item) => {
-          const name = item.originalContent?.properties?.referenceName || "Unknown";
-          const contentID = item.originalContent?.contentID;
-          allFailureDetails.push({
-            name,
-            error: item.error,
-            type: "content",
-            contentID,
-            guid: sourceGuidStr,
-            locale,
-          });
-          // Register in global registry so page pusher can provide better error messages
-          if (contentID) {
-            registerFailedContent(contentID, name, item.error, locale);
+          const created = new Set((result.successfulItems || []).map((item) => item.originalContent?.contentID));
+          const toComplete = stubbed.filter((item) => created.has(item.contentID));
+          if (toComplete.length > 0) {
+            const completed = await cycleBatchProcessor.processBatches(toComplete, logger, `${label}: completing reference cycles`);
+            recordBatchResult(completed);
+            // Saved without its references: make the next sync push it again
+            for (const failed of completed.failedItems || []) {
+              const contentID = failed.originalContent?.contentID;
+              if (contentID) referenceMapper.markSourceChanged(contentID);
+            }
           }
-        });
-      }
+        }
       }
     }
 
@@ -202,37 +232,15 @@ export async function pushContent(sourceData: ContentItem[], targetData: Content
         totalSuccessful += filteredNormalContentItems.itemsToProcess.length;
         totalSkipped += filteredNormalContentItems.skippedCount;
       } else {
-      const normalBatchProcessor = new ContentBatchProcessor(normalBatchConfig);
-      const normalResult = await normalBatchProcessor.processBatches(
-        filteredNormalContentItems.itemsToProcess as ContentItem[],
-        logger,
-        "Normal Content"
-      );
+        const normalBatchProcessor = new ContentBatchProcessor(normalBatchConfig);
+        const normalResult = await normalBatchProcessor.processBatches(
+          filteredNormalContentItems.itemsToProcess as ContentItem[],
+          logger,
+          "Normal Content"
+        );
 
-      totalSuccessful += normalResult.successCount;
-      totalFailed += normalResult.failureCount;
-      totalSkipped += filteredNormalContentItems.skippedCount;
-      totalSkipped += normalResult.skippedCount;
-      allPublishableIds.push(...normalResult.publishableIds);
-      // Collect failure details for error summary and register in failed content registry
-      if (normalResult.failedItems && normalResult.failedItems.length > 0) {
-        normalResult.failedItems.forEach((item) => {
-          const name = item.originalContent?.properties?.referenceName || "Unknown";
-          const contentID = item.originalContent?.contentID;
-          allFailureDetails.push({
-            name,
-            error: item.error,
-            type: "content",
-            contentID,
-            guid: sourceGuidStr,
-            locale,
-          });
-          // Register in global registry so page pusher can provide better error messages
-          if (contentID) {
-            registerFailedContent(contentID, name, item.error, locale);
-          }
-        });
-      }
+        totalSkipped += filteredNormalContentItems.skippedCount;
+        recordBatchResult(normalResult);
       }
     }
 
@@ -257,3 +265,62 @@ export async function pushContent(sourceData: ContentItem[], targetData: Content
     };
   }
 }
+
+/**
+ * Groups the items left after filtering by their dependency level (PROD-3025), dropping levels
+ * the filter emptied. An item missing from every level goes last, after everything it could need.
+ */
+export function splitIntoLevels(items: ContentItem[], levels: ContentItem[][]): ContentItem[][] {
+  const levelById = new Map<number, number>();
+  levels.forEach((level, i) => level.forEach((item) => levelById.set(item.contentID, i)));
+
+  const grouped: ContentItem[][] = levels.map(() => []);
+  const unplaced: ContentItem[] = [];
+  for (const item of items) {
+    const level = levelById.get(item.contentID);
+    if (level === undefined) unplaced.push(item);
+    else grouped[level].push(item);
+  }
+  grouped.push(unplaced);
+  return grouped.filter((level) => level.length > 0);
+}
+
+/**
+ * PROD-3076: for the items of one level, the copies to push in pass 1. An item in a reference cycle
+ * whose cycle partners (or itself) have no target ID yet goes out with those references blanked,
+ * and is returned in `stubbed` for pass 2. Everything else is pushed as-is. Partners that already
+ * have a mapping (a re-sync) are left alone, so a re-sync never blanks anything.
+ */
+export function stubCycleReferences(
+  items: ContentItem[],
+  cycleReferences: Map<number, number[]>,
+  referenceMapper: ContentItemMapper,
+  modelMapper: ModelMapper
+): { toPush: ContentItem[]; stubbed: ContentItem[] } {
+  const toPush: ContentItem[] = [];
+  const stubbed: ContentItem[] = [];
+
+  for (const item of items) {
+    const unmapped = (cycleReferences.get(item.contentID) || []).filter(
+      (id) => !referenceMapper.getContentItemMappingByContentID(id, "source")
+    );
+    if (unmapped.length === 0) {
+      toPush.push(item);
+      continue;
+    }
+
+    const definitionName = item.properties?.definitionName?.toLowerCase();
+    const modelMapping = definitionName ? modelMapper.getModelMappingByReferenceName(definitionName, "source") : null;
+    const sourceModel = modelMapping ? modelMapper.getMappedEntity(modelMapping, "source") : null;
+    const { fields, stripped } = stripContentReferences(item.fields, new Set(unmapped), sourceModel);
+    if (stripped === 0) {
+      toPush.push(item);
+      continue;
+    }
+
+    toPush.push({ ...item, fields });
+    stubbed.push(item);
+  }
+
+  return { toPush, stubbed };
+}