Price Update Streaming Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Stop the scheduled price update from materialising every variant in a store, so large stores stop OOM-killing the worker.

Architecture: Phase 1 (fetchAllProductsWithPagination) becomes an async generator (streamManagedVariants). Phase 2 (calculatePriceUpdates) consumes it lazily, accumulating variants into per-game buffers that flush at VARIANT_BATCH_SIZE, with an aggregate cap across all games so peak memory is independent of both store size and game count. Phases 3 and 4 are untouched.

Tech Stack: Node 24 CommonJS, Vitest (test files use ESM import), Mongoose, BullMQ, Shopify GraphQL.

Spec: docs/superpowers/specs/2026-09-03-price-update-streaming-design.md

Global Constraints

Never vi.mock a server module. vitest.config.js externalises server source to native require(), so such a mock is silently inert and the test reaches the real dependency. Use the existing _setDeps() / _resetDeps() seam. (CLAUDE.md §3)
No identity defaults. No || 'mtg', ?? 'mtg', or any fallback for game, shop, or a pricing config. Absent → throw. (CLAUDE.md §5.5)
Do not mutate caller-supplied arrays or Maps. Buffers are allocated and owned internally. (CLAUDE.md §5.11)
Coverage: every modified file must report ≥70% on all four metrics. Read the per-file table — the Vitest 4 threshold gate is silently inoperative and the command exits 0 regardless. (CLAUDE.md §7)
Commits: one imperative sentence stating the merchant-visible outcome, sentence case, no trailing period. End every commit message with Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>.
Never --no-verify. Husky runs ESLint pre-commit; fix lint rather than bypassing.
Existing constant, do not change: VARIANT_BATCH_SIZE = parseInt(process.env.PRICE_UPDATE_VARIANT_BATCH_SIZE, 10) || 5000 at server/services/priceUpdateService.js:38.
priceUpdateStats schema and the client are frozen in this plan. No new stats fields, no client/ changes.

File Structure

File	Responsibility	Action
`server/services/priceUpdateService.js`	Phase 1 pagination + phase 2 calculation	Modify
`server/services/priceUpdateService.test.js`	Co-located tests for the above	Modify
`server/queues/processors/priceUpdateProcessor.js`	Phase orchestration and `SyncJob` stats writes	Modify
`server/queues/processors/priceUpdateProcessor.test.js`	Co-located processor tests	Modify
`CLAUDE.md`	§5.6 status line	Modify (Task 5)
`GCE_WORKER_DEPLOYMENT.md`	Memory note at line 329	Modify (Task 5)

No new files. The change is confined to two source files and their co-located tests.

Task 1: Golden equivalence harness

Pins the current implementation's exact output before anything changes. This must land first — a golden captured after the rewrite only pins the rewrite's own behaviour.

Files:

Test: server/services/priceUpdateService.test.js

Interfaces:

Consumes: the existing calculatePriceUpdates(variantMap, pricingConfigs, onProgress, options) signature.
Produces: buildMixedGameFixture() and GOLDEN_RESULT, used by Task 3 to prove equivalence.
[ ] Step 1: Read the existing test file's setup

Run: sed -n '1,80p' server/services/priceUpdateService.test.js

Note how it constructs a PriceUpdateService, what it injects via _setDeps, and how it fakes plugins and catalog models. Reuse that harness verbatim rather than inventing a second one.

[ ] Step 2: Add the fixture builder and the golden test

Add to server/services/priceUpdateService.test.js:

// A deterministic mixed-game variant set. Kept small enough to assert on by
// hand, wide enough to exercise every counter: two games, a price-locked
// product, an unmanaged product, a DRAFT product, and a SKU with no catalog
// match (the noPriceData path).
function buildMixedGameFixture() {
    const variants = [];
    const push = (i, game, extra = {}) => variants.push({
        sku: `${game.toUpperCase()}-${i}`,
        variantId: `gid://shopify/ProductVariant/${game}-${i}`,
        productId: `gid://shopify/Product/${game}-${i}`,
        productStatus: 'ACTIVE',
        currentPrice: 1.00,
        productHandle: `${game}-card-${i}`,
        setCode: 'set1',
        rarity: 'common',
        game,
        boosterGame: game,
        managedBy: 'lgs-forge',
        priceLocked: false,
        finish: 'Normal',
        condition: null,
        ...extra
    });

    for (let i = 0; i < 12; i++) push(i, 'mtg');
    for (let i = 0; i < 8; i++) push(i, 'pokemon');
    push(99, 'mtg', { priceLocked: true });
    push(98, 'mtg', { game: 'unmanaged', boosterGame: null, managedBy: null });
    push(97, 'pokemon', { productStatus: 'DRAFT' });
    push(96, 'mtg', { sku: 'MTG-NO-CATALOG-MATCH' });

    return new Map(variants.map(v => [v.variantId, v]));
}

describe('calculatePriceUpdates — golden equivalence', () => {
    it('produces the recorded result for the mixed-game fixture', async () => {
        const service = makeService();           // from the existing harness
        const fixture = buildMixedGameFixture();

        const result = await service.calculatePriceUpdates(
            fixture,
            { mtg: MTG_PRICING_CONFIG, pokemon: POKEMON_PRICING_CONFIG }
        );

        expect({
            updates: result.updates,
            productsToActivate: result.productsToActivate,
            pricesMatched: result.pricesMatched,
            noPriceData: result.noPriceData,
            unmanaged: result.unmanaged,
            priceLocked: result.priceLocked
        }).toMatchSnapshot();
    });
});

makeService, MTG_PRICING_CONFIG and POKEMON_PRICING_CONFIG are whatever the existing file already calls these things — reuse its names, do not add duplicates.

[ ] Step 3: Run it to record the snapshot

Run: npx vitest run server/services/priceUpdateService.test.js -t "golden equivalence" Expected: PASS, and a new snapshot written to server/services/__snapshots__/priceUpdateService.test.js.snap.

[ ] Step 4: Verify the snapshot is non-trivial

Run: grep -c "newPrice" server/services/__snapshots__/priceUpdateService.test.js.snap Expected: a non-zero count. A snapshot of an empty updates array proves nothing — if it is empty, the fake catalog model is not returning products for these SKUs. Fix the fixture to match the harness's catalog stub before continuing.

[ ] Step 5: Commit

bash

git add server/services/priceUpdateService.test.js server/services/__snapshots__/priceUpdateService.test.js.snap
git commit -m "$(cat <<'EOF'
Record the price calculation's current output before restructuring it

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
EOF
)"

Task 2: Extract the per-batch body (pure refactor)

Lifts the innards of the nested for game { for offset { ... } } loops into one inner closure. This exists so Task 3's diff is small and reviewable.

One deliberate behaviour change, ruled on during pre-flight: the missing-pricingConfig error currently reads Missing pricingConfig for game '${gameId}' (${gameVariants.length} variants to price) (priceUpdateService.js:511). contextFor drops the parenthetical, because once Task 3 streams there is no variant count to report at throw time. priceUpdateService.test.js:182 asserts only the leading fragment via toThrow(string), which is a substring match, so that test still passes. Everything else in this task is behaviour-preserving.

Files:

Modify: server/services/priceUpdateService.js:395-676

Interfaces:

Consumes: Task 1's golden snapshot as the regression gate.
Produces: an inner closure priceBatch(gameId, batch) — async (string, Array<object>) => void — that Task 3 calls from the streaming loop. It is a closure, not a class method, so it captures the counters (variantsChecked, pricesMatched, noPriceData, draftProductsWithPricing) directly instead of threading a mutable context object.
[ ] Step 1: Add a per-game memo above the game loop

In calculatePriceUpdates, immediately after the fireProgress helper definition, add:

// Per-game constants resolved once and reused by every batch of that game.
// getPlugin and getPriceCalculationProjection are pure, but the pricingConfig
// check must fire before any pricing work — see §5.5, no defaults.
const gameContext = new Map();
const contextFor = (gameId) => {
    if (gameContext.has(gameId)) return gameContext.get(gameId);
    const plugin = getPlugin(gameId);
    const pricingConfig = pricingConfigs[gameId];
    if (!pricingConfig) {
        throw new Error(`Missing pricingConfig for game '${gameId}'`);
    }
    const ctx = { plugin, pricingConfig, projection: plugin.getPriceCalculationProjection() };
    gameContext.set(gameId, ctx);
    return ctx;
};

[ ] Step 2: Convert the inner loop body into priceBatch

Replace the whole for (const [gameId, gameVariants] of variantsByGame) { ... } block with the closure plus a loop that calls it. The closure body is the existing contents of the for (let offset = ...) block, verbatim, with three substitutions: plugin → ctx.plugin, projection → ctx.projection, and the logger.info('Pre-loading prices', ...) call drops its now-meaningless batchOffset and gameVariants fields.

const priceBatch = async (gameId, batch) => {
    const ctx = contextFor(gameId);
    const emitPreload = onProgress
        ? async (stage, preload) => onProgress({
            variantsChecked,
            totalVariants,
            pricesNeedUpdate: updates.length,
            pricesMatched,
            noPriceData,
            unmanaged,
            draftProductsWithPricing,
            stage: `${gameId}_preload_${stage}`,
            preload
        })
        : null;

    const skus = batch.map(v => v.sku).filter(Boolean);
    const catalogProducts = await _getCatalogModel(ctx.plugin).find({
        $or: [
            { 'variants.sku': { $in: skus } },
            { sku: { $in: skus } }
        ]
    }).select(ctx.projection).lean();

    // ... the rest of the existing offset-loop body, unchanged:
    //     skuToProduct, cardIds, emitPreload('mongo_fetch_start'),
    //     ctx.plugin.preloadPrices(...), emitPreload('complete'),
    //     for (const variantData of batch) { ... }
};

for (const [gameId, gameVariants] of variantsByGame) {
    for (let offset = 0; offset < gameVariants.length; offset += batchSize) {
        await priceBatch(gameId, gameVariants.slice(offset, offset + batchSize));
    }
}

[ ] Step 3: Run the full service test file

Run: npx vitest run server/services/priceUpdateService.test.js Expected: PASS, including the golden snapshot with no -u flag. A snapshot mismatch here means the extraction changed behaviour — fix the extraction, never update the snapshot.

[ ] Step 4: Lint

Run: npm run lint Expected: exit 0, no new warnings.

[ ] Step 5: Commit

bash

git add server/services/priceUpdateService.js
git commit -m "$(cat <<'EOF'
Extract the price calculation's per-batch body so it can be driven by a stream

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
EOF
)"

Task 3: Buffer per game and consume an async iterable

The core change. calculatePriceUpdates stops receiving a materialised Map and starts consuming any async iterable, buffering per game with an aggregate cap.

Files:

Modify: server/services/priceUpdateService.js:38 (new constant), :395-676
Test: server/services/priceUpdateService.test.js

Interfaces:

Consumes: priceBatch(gameId, batch) from Task 2.
Produces: calculatePriceUpdates(variantSource, pricingConfigs, onProgress, options) where variantSource is Map<string, object> | Iterable<object> | AsyncIterable<object>. Return value gains variantsSeen: number; all existing fields (updates, productsToActivate, pricesMatched, noPriceData, unmanaged, priceLocked) keep their names and types. Task 4 consumes variantsSeen.
[ ] Step 1: Write the failing tests

Add to server/services/priceUpdateService.test.js:

describe('calculatePriceUpdates — bounded buffering', () => {
    afterEach(() => service._resetDeps?.());

    // Yields `count` variants alternating across `games`, recording how many
    // have been yielded so far so a test can prove consumption is lazy.
    function makeStream(count, games, state = {}) {
        state.yielded = 0;
        return (async function* () {
            for (let i = 0; i < count; i++) {
                const game = games[i % games.length];
                state.yielded++;
                yield {
                    sku: `${game}-${i}`,
                    variantId: `gid://v/${i}`,
                    productId: `gid://p/${i}`,
                    productStatus: 'ACTIVE',
                    currentPrice: 1.00,
                    productHandle: `h-${i}`,
                    setCode: 'set1',
                    rarity: 'common',
                    game,
                    boosterGame: game,
                    managedBy: 'lgs-forge',
                    priceLocked: false,
                    finish: 'Normal',
                    condition: null
                };
            }
        })();
    }

    it('never buffers more than maxBufferedVariants', async () => {
        const seen = [];
        setDeps({ variantBatchSize: 10, maxBufferedVariants: 40,
                  onBufferSample: (total) => seen.push(total) });

        await makeService().calculatePriceUpdates(
            makeStream(5000, ['mtg', 'pokemon']), ALL_CONFIGS
        );

        expect(Math.max(...seen)).toBeLessThanOrEqual(40);
    });

    it('respects the aggregate cap when more games are present than the cap allows', async () => {
        const seen = [];
        setDeps({ variantBatchSize: 10, maxBufferedVariants: 25,
                  onBufferSample: (total) => seen.push(total) });

        // 8 games x batchSize 10 = 80 > cap 25, so the cap must bite.
        await makeService().calculatePriceUpdates(
            makeStream(4000, ['mtg', 'pokemon', 'riftbound', 'lorcana',
                              'mtg', 'pokemon', 'riftbound', 'lorcana']),
            ALL_CONFIGS
        );

        expect(Math.max(...seen)).toBeLessThanOrEqual(25);
    });

    it('starts pricing before the source is exhausted', async () => {
        const state = {};
        let yieldedAtFirstBatch = null;
        setDeps({ variantBatchSize: 10, maxBufferedVariants: 40,
                  onBatchStart: () => {
                      if (yieldedAtFirstBatch === null) yieldedAtFirstBatch = state.yielded;
                  } });

        await makeService().calculatePriceUpdates(
            makeStream(5000, ['mtg'], state), ALL_CONFIGS
        );

        expect(yieldedAtFirstBatch).toBeLessThan(5000);
    });

    it('reports variantsSeen for every variant including skipped ones', async () => {
        setDeps({ variantBatchSize: 10, maxBufferedVariants: 40 });
        const result = await makeService().calculatePriceUpdates(
            makeStream(100, ['mtg']), ALL_CONFIGS
        );
        expect(result.variantsSeen).toBe(100);
    });

    it('produces the same per-game batch sizes as group-then-batch did', async () => {
        const batches = [];
        setDeps({ variantBatchSize: 10, maxBufferedVariants: 40,
                  onBatchStart: (gameId, size) => batches.push([gameId, size]) });

        await makeService().calculatePriceUpdates(
            makeStream(100, ['mtg', 'pokemon']), ALL_CONFIGS
        );

        // 50 of each game at batchSize 10 => five full batches per game, no remainder.
        expect(batches.filter(([g]) => g === 'mtg')).toEqual(Array(5).fill(['mtg', 10]));
        expect(batches.filter(([g]) => g === 'pokemon')).toEqual(Array(5).fill(['pokemon', 10]));
    });

    it('throws on a missing pricingConfig at the first variant of that game, not after draining', async () => {
        const state = {};
        setDeps({ variantBatchSize: 10, maxBufferedVariants: 40 });

        await expect(
            makeService().calculatePriceUpdates(
                makeStream(5000, ['mtg'], state), { pokemon: POKEMON_PRICING_CONFIG }
            )
        ).rejects.toThrow("Missing pricingConfig for game 'mtg'");

        expect(state.yielded).toBeLessThan(5000);
    });
});

setDeps and ALL_CONFIGS follow the existing file's naming — reuse, don't duplicate.

[ ] Step 2: Run the tests to verify they fail

Run: npx vitest run server/services/priceUpdateService.test.js -t "bounded buffering" Expected: FAIL. maxBufferedVariants, onBufferSample, onBatchStart and variantsSeen do not exist yet.

[ ] Step 3: Add the aggregate-cap constant

At server/services/priceUpdateService.js, directly below the existing VARIANT_BATCH_SIZE on line 38:

// Peak buffered variants across ALL games. Per-game buffers alone would make
// peak scale with the number of registered games — 100 games x 5000 would
// reinstate the store-sized heap this streaming rewrite removes. The default
// preserves current behaviour exactly for any store carrying four games or
// fewer, which is every store today.
const MAX_BUFFERED_VARIANTS = parseInt(process.env.PRICE_UPDATE_MAX_BUFFERED_VARIANTS, 10)
    || (VARIANT_BATCH_SIZE * 4);

[ ] Step 4: Replace grouping with streaming

In calculatePriceUpdates, delete the variantsByGame construction loop (the for (const variantData of variantMap.values()) block) and the for (const [gameId, gameVariants] of variantsByGame) driver from Task 2. Rename the first parameter to variantSource. Insert after the priceBatch definition:

const maxBuffered = (_deps && _deps.maxBufferedVariants) || MAX_BUFFERED_VARIANTS;

// gameId -> variantData[]. Allocated and owned here; no caller array is ever
// mutated (§5.11).
const buffers = new Map();
const seenByGame = new Map();
let totalBuffered = 0;
let variantsSeen = 0;

const flushGame = async (gameId) => {
    const batch = buffers.get(gameId);
    if (!batch || batch.length === 0) return;
    buffers.set(gameId, []);
    totalBuffered -= batch.length;
    if (_deps && _deps.onBatchStart) _deps.onBatchStart(gameId, batch.length);
    await priceBatch(gameId, batch);
};

// Flushing the largest buffer keeps batches as close to VARIANT_BATCH_SIZE as
// the cap allows, so batch efficiency degrades gracefully rather than sharply.
const flushLargest = async () => {
    let victim = null;
    let victimLength = 0;
    for (const [gameId, buf] of buffers) {
        if (buf.length > victimLength) { victim = gameId; victimLength = buf.length; }
    }
    if (victim) await flushGame(victim);
};

// `for await` consumes sync iterables too, so a Map's values, a plain array
// and an async generator all work — existing callers and fixtures unchanged.
const source = variantSource instanceof Map ? variantSource.values() : variantSource;

for await (const variantData of source) {
    variantsSeen++;

    // Checked before the game lookup: a locked product is one the merchant
    // asked us to leave alone, whether or not we can price it.
    if (variantData.priceLocked) {
        priceLocked++;
        variantsChecked++;
        continue;
    }
    if (!_pluginForGame(variantData.game)) {
        unmanaged++;
        variantsChecked++;
        continue;
    }

    if (!buffers.has(variantData.game)) {
        contextFor(variantData.game);   // throws now, not after draining the store
        buffers.set(variantData.game, []);
    }
    buffers.get(variantData.game).push(variantData);
    totalBuffered++;
    seenByGame.set(variantData.game, (seenByGame.get(variantData.game) || 0) + 1);
    if (_deps && _deps.onBufferSample) _deps.onBufferSample(totalBuffered);

    if (buffers.get(variantData.game).length >= batchSize) {
        await flushGame(variantData.game);
    } else if (totalBuffered >= maxBuffered) {
        await flushLargest();
    }
}

for (const gameId of [...buffers.keys()]) {
    await flushGame(gameId);
}

logger.info('Variant breakdown by game', {
    shop: this.shop,
    ...Object.fromEntries(seenByGame),
    unmanaged,
    priceLocked
});

[ ] Step 5: Make totalVariants track the running count

totalVariants was variantMap.size, known upfront. Delete that binding. In fireProgress, in priceBatch's emitPreload, and in the final onProgress call, replace totalVariants with totalVariants: variantsSeen. The progress payload keeps the same field name, so priceUpdateStats and the client are untouched.

[ ] Step 6: Return the new field

return {
    updates,
    productsToActivate: Array.from(productsToActivate.values()),
    pricesMatched,
    noPriceData,
    unmanaged,
    priceLocked,
    variantsSeen
};

[ ] Step 7: Run the whole service test file

Run: npx vitest run server/services/priceUpdateService.test.js Expected: PASS, all tests, including Task 1's golden snapshot with no -u. The golden passes a Map, which the instanceof Map branch still accepts. If the snapshot now differs, the streaming rewrite changed behaviour — that is the bug this task exists to avoid. Fix the code, never the snapshot.

[ ] Step 8: Commit

bash

git add server/services/priceUpdateService.js server/services/priceUpdateService.test.js
git commit -m "$(cat <<'EOF'
Price variants from a bounded buffer instead of the whole store at once

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
EOF
)"

Task 4: Stream phase 1 and wire the processor

Turns phase 1 into a generator and connects it. This is the task that actually fixes the OOM — until now phase 1 still materialised.

Files:

Modify: server/services/priceUpdateService.js:214-378
Modify: server/queues/processors/priceUpdateProcessor.js:207-283
Test: server/queues/processors/priceUpdateProcessor.test.js

Interfaces:

Consumes: calculatePriceUpdates(variantSource, ...) and result.variantsSeen from Task 3.
Produces: streamManagedVariants(collectionId, onProgress) — an async generator yielding variant-data objects one at a time, with the same per-page onProgress({ pageCount, totalProducts, totalVariants }) contract fetchAllProductsWithPagination had. Throws MANAGED_COLLECTION_MISSING from the first page as before.
Also produces: fetchAllProductsWithPagination(onProgress, collectionId) retained as a thin wrapper that drains the generator into a Map. See the pre-flight ruling below — this is not optional.

Pre-flight ruling — do not skip. marketplacePushProcessor.js:360 and :371 also call priceService.fetchAllProductsWithPagination(null, collectionId) and use the result as a Map, and four tests in marketplacePushProcessor.test.js mock that name. Renaming outright would break the marketplace push queue. Keep the old method as a wrapper so that path is untouched:
js
// Retained for marketplacePushProcessor, which needs the whole map at once.
// The price-update path deliberately does NOT use this — draining the
// generator here reinstates exactly the store-sized heap that
// streamManagedVariants exists to avoid.
async fetchAllProductsWithPagination(onProgress, collectionId) {
    const variantMap = new Map();
    for await (const variant of this.streamManagedVariants(collectionId, onProgress)) {
        variantMap.set(variant.variantId, variant);
    }
    return variantMap;
}
1
2
3
4
5
6
7
8
9
10
11
Converting marketplace push to streaming is out of scope for this plan.

[ ] Step 1: Write the failing processor test

Add to server/queues/processors/priceUpdateProcessor.test.js:

it('recreates the collection and restarts cleanly when it is missing', async () => {
    let attempt = 0;
    const streamManagedVariants = vi.fn(async function* () {
        attempt++;
        if (attempt === 1) {
            const err = new Error('Managed collection missing or inaccessible.');
            err.code = 'MANAGED_COLLECTION_MISSING';
            throw err;
        }
        yield makeVariant('mtg', 1);
        yield makeVariant('mtg', 2);
    });

    setDeps({ PriceUpdateService: fakeServiceClass({ streamManagedVariants }),
              ensureManagedCollection: vi.fn(async () => 'gid://shopify/Collection/new') });

    const result = await priceUpdateProcessor(makeJob());

    expect(attempt).toBe(2);
    // The failed attempt must not leak counters into the successful one.
    expect(result.variantsChecked).toBe(2);
});

[ ] Step 2: Run it to verify it fails

Run: npx vitest run server/queues/processors/priceUpdateProcessor.test.js -t "recreates the collection" Expected: FAIL — streamManagedVariants is not called; the processor still calls fetchAllProductsWithPagination.

[ ] Step 3: Convert phase 1 to a generator

In server/services/priceUpdateService.js, rename fetchAllProductsWithPagination to streamManagedVariants and change async fetchAllProductsWithPagination(onProgress, collectionId) to async *streamManagedVariants(collectionId, onProgress) — note the reordered parameters, so the required argument comes first.

Then, inside the product loop, replace the variantMap.set(variant.id, {...}) call with a yield of the same object literal, and delete the variantMap declaration and the return variantMap. Replace the three variantMap.size references with a local variantsYielded counter incremented at each yield, so the onProgress and logger.success payloads keep their existing field names.

if (sku && variant.id) {
    variantsYielded++;
    yield {
        sku,
        variantId: variant.id,
        productId: productContext.productId,
        productStatus: productContext.productStatus,
        currentPrice,
        productHandle: productContext.productHandle,
        setCode: productContext.setCode,
        rarity: productContext.rarity,
        game: productContext.game,
        boosterGame: productContext.boosterGame,
        managedBy: productContext.managedBy,
        priceLocked: productContext.priceLocked,
        finish,
        condition
    };
}

[ ] Step 4: Rewrite the processor's phases 1–2

In server/queues/processors/priceUpdateProcessor.js, replace everything from let variantMap; through the updatePriceStatsAtomic call that sets 'variantsChecked': variantMap.size with:

// One attempt = one fresh generator AND one fresh calculatePriceUpdates call,
// so a MANAGED_COLLECTION_MISSING retry cannot leak counters or half-built
// updates from the failed attempt into the successful one.
const runPipeline = async (collectionId) => {
    let variantsYielded = 0;

    // Phase 1 now completes only when the generator is exhausted, which happens
    // inside calculatePriceUpdates. Wrapping it is how the processor still owns
    // the phase-1 stats write.
    const trackedStream = (async function* () {
        for await (const variant of priceService.streamManagedVariants(collectionId, onPhase1Progress)) {
            variantsYielded++;
            yield variant;
        }
        await updatePriceStatsAtomic(syncJobId, {
            'totalVariants': variantsYielded,
            'pipeline.phase1.status': 'completed',
            'pipeline.phase1.completedAt': new Date(),
            'pipeline.phase1.progress.totalProducts': variantsYielded
        });
        // Kept from the pre-streaming flow: phase 1 ending is still the 30% mark.
        await job.updateProgress(30);
    })();

    await updatePriceStatsAtomic(syncJobId, { 'pipeline.phase2.status': 'running' });

    return priceService.calculatePriceUpdates(trackedStream, pricingConfigs, async (progress) => {
        await updatePriceStatsAtomic(syncJobId, {
            'pipeline.phase2.progress.variantsChecked': progress.variantsChecked,
            'pipeline.phase2.progress.totalVariants': progress.totalVariants,
            'pipeline.phase2.progress.pricesNeedUpdate': progress.pricesNeedUpdate,
            'pipeline.phase2.progress.pricesMatched': progress.pricesMatched,
            'pipeline.phase2.progress.noPriceData': progress.noPriceData,
            'pipeline.phase2.progress.draftProductsWithPricing': progress.draftProductsWithPricing,
            'pipeline.phase2.progress.stage': progress.stage || null,
            'pipeline.phase2.progress.preload': progress.preload || null
        });
    });
};

let priceResult;
try {
    priceResult = await runPipeline(managedCollectionId);
} catch (err) {
    if (err.code === 'MANAGED_COLLECTION_MISSING') {
        logger.warn('Managed collection missing on Shopify; clearing cache and recreating', {
            shop, staleCollectionId: managedCollectionId
        });
        await _store().updateOne({ shop }, { $set: { managedCollectionId: null } });
        const recreatedId = await _ensureManagedCollection()(shop, accessToken);
        // Reset any phase-2 progress the failed attempt wrote, so the merchant
        // does not see counters from a run that produced nothing.
        await updatePriceStatsAtomic(syncJobId, { 'pipeline.phase2.progress': {} });
        priceResult = await runPipeline(recreatedId);
    } else {
        throw err;
    }
}

await job.updateProgress(50);
await checkIfCancelled(syncJobId);

Then replace the four former variantMap.size reads (the 'variantsChecked' stat write and the two logger.info payloads) with priceResult.variantsSeen.

[ ] Step 5: Add the per-flush cancellation check

Phases 1 and 2 are now one long stretch, so the between-phase checkIfCancelled no longer bounds how long a cancelled job keeps working. In server/services/priceUpdateService.js, in flushGame, immediately before await priceBatch(gameId, batch):

if (options.onBeforeBatch) await options.onBeforeBatch();

and in the processor's runPipeline, pass it through:

return priceService.calculatePriceUpdates(trackedStream, pricingConfigs, onPhase2Progress, {
    onBeforeBatch: () => checkIfCancelled(syncJobId)
});

[ ] Step 6: Run both test files

Run: npx vitest run server/services/priceUpdateService.test.js server/queues/processors/priceUpdateProcessor.test.js Expected: PASS, including the golden snapshot.

[ ] Step 6b: Update the four existing processor tests

server/queues/processors/priceUpdateProcessor.test.js:53, :85, :128, :263 each stub this.fetchAllProductsWithPagination = vi.fn().mockResolvedValue(<Map>). The processor no longer calls that method, so these stubs go unused and the tests fail. Convert each to stub the generator instead, preserving the same variants:

// was: this.fetchAllProductsWithPagination = vi.fn().mockResolvedValue(variantMap);
this.streamManagedVariants = vi.fn(async function* () {
    for (const v of variantMap.values()) yield v;
});

Leave marketplacePushProcessor.test.js alone — it exercises the retained wrapper and must keep passing unchanged. That it still passes is the proof the wrapper works.

[ ] Step 7: Confirm every remaining caller of the old name is intentional

Run: grep -rn "fetchAllProductsWithPagination" --include=*.js server/ client/

Expected: exactly these, and nothing else —

the wrapper's own definition in priceUpdateService.js
marketplacePushProcessor.js:360 and :371
the four marketplacePushProcessor.test.js stubs
JSDoc mentions in BaseMarketplaceAdapter.js, cardtrader/mapping.js, shopifyAPI.js

A hit in priceUpdateProcessor.js means this task is incomplete — that is the path that must not materialise.

[ ] Step 8: Commit

bash

git add server/services/priceUpdateService.js server/queues/processors/priceUpdateProcessor.js server/queues/processors/priceUpdateProcessor.test.js
git commit -m "$(cat <<'EOF'
Stream a store's variants through the price update instead of loading them all

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
EOF
)"

Task 5: Full verification and documentation

Files:

Modify: CLAUDE.md (§5.6 status paragraph)
Modify: GCE_WORKER_DEPLOYMENT.md:329
[ ] Step 1: Run the full server suite

Run: npm test Expected: exit 0.

If ~49 files fail on missing client deps or a Mongo/Redis connection, that is the known fresh-worktree condition, not a regression: run npm run install:all and docker compose up -d, then re-run.

[ ] Step 2: Check per-file coverage

Run: npm run test:coverage

Read the per-file table for priceUpdateService.js and priceUpdateProcessor.js. Both must show ≥70% on statements, branches, functions and lines. The command exits 0 regardless — the configured threshold gate is inoperative under Vitest 4, so the table is the only signal.

[ ] Step 3: Lint

Run: npm run lint Expected: exit 0 with no new warnings.

[ ] Step 4: Grep the diff for the failure modes this repo names

bash

git diff main --unified=0 -- server/ | grep -nE "\|\| *'mtg'|\?\? *'mtg'|= *'mtg'" || echo "no identity defaults - good"
git diff main --unified=0 -- server/ | grep -nE "myshopify\.com" || echo "no direct Shopify hosts - good"
git diff main --unified=0 -- server/ | grep -nE "\.splice\(|\.sort\(|length = 0" || echo "no in-place param mutation - good"

Expected: all three print their "good" line. A .sort( hit inside a function you own on a buffer you allocated is fine — confirm it is not on a parameter.

[ ] Step 5: Update CLAUDE.md §5.6

The status paragraph ends by naming the MTG path as streaming through loadLatestSnapshotsBySlug. Append:

markdown

The price-update path's *phase 1* was a separate instance of the same shape and is
now also streamed: `priceUpdateService.streamManagedVariants` yields variants and
`calculatePriceUpdates` buffers them per game under an aggregate
`MAX_BUFFERED_VARIANTS` cap, so peak heap tracks the cap rather than the store's
variant count. Before that, `alchemists-refuge` at 764,806 variants OOM-killed the
2,560 MB worker every ~2 hours (2026-09-03), and because the crash starved the
`WorkerHeartbeat` write, every failed row blamed the worker for "not responding"
rather than naming the heap.

[ ] Step 6: Update GCE_WORKER_DEPLOYMENT.md

Line 329 currently reads:

Do not downsize to e2-small without first bounding peak memory in the price-update path.

Replace with:

The price-update path's peak memory is now bounded (MAX_BUFFERED_VARIANTS in priceUpdateService.js), so it no longer scales with a store's variant count. Still measure before downsizing: phase 3's updates[] and the co-resident Redis both remain unbounded by this change.

[ ] Step 7: Commit

bash

git add CLAUDE.md GCE_WORKER_DEPLOYMENT.md
git commit -m "$(cat <<'EOF'
Record that the price update's peak memory no longer scales with store size

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
EOF
)"

[ ] Step 8: Verify against the real store before re-enabling its schedule

Unit tests cannot prove the heap ceiling is cleared. alchemists-refuge is the only known store at this scale, so it is the only meaningful test.

With the schedule still paused, trigger one manual price update for that shop and watch the worker's RSS. Expected: the run completes, and RSS stays well below 2,560 MB rather than climbing toward it.

bash

gcloud compute ssh lgs-ledger-worker --command \
  "watch -n 30 'ps -o rss=,etime=,cmd= -C node | grep worker.js'"

Do not re-enable the schedule until this run completes. Re-enabling is the merchant-facing step and belongs to Brent, not to this plan.

Self-Review

Spec coverage:

Spec section	Task
§3.1 streaming shape	4
§3.2 per-game buffers	3
§3.2.1 aggregate cap	3 (constant, `flushLargest`, cap test)
§3.3 interface change, `variantsSeen`	3 (service), 4 (processor)
§3.5 phases 3–4 untouched	no task — verified by Task 5 step 7's absence of diff there
§4 merchant-visible phases	4 step 4 — phase 1 stats move into the wrapper; no client change
§5.1 retry resets accumulators	4 steps 1, 4
§5.2 cancellation per flush	4 step 5
§5.3 §5.11 aliased array	3 step 4 (`buffers` owned internally); Task 5 step 4 greps for it
§5.4 §5.1 game parity	3 (mixed-game and 8-game tests exercise every registered plugin path)
§5.5 no identity defaults	3 step 4 (`contextFor` throws); Task 5 step 4 greps for it
§6 tests 1, 1b, 2, 3, 4, 5, 6	1 (golden/4), 3 (1, 1b, 2, batch composition), 4 (5, 6)
§6.1 manual verification	5 step 8
§7 deferred	no task, by design
§8 Redis	out of scope, separate spec
§9 definition of done	5

Gap found and closed: spec §6 test 3 ("batch composition unchanged") initially had no task — Task 1's golden fixture is only 20 variants and never reaches a batch boundary, so nothing would have caught a change in batch sizing. A produces the same per-game batch sizes as group-then-batch did test is now in Task 3, Step 1.

Placeholder scan: no TBD/TODO; every code step carries real code; the one "the rest of the existing offset-loop body, unchanged" in Task 2 names the exact statements to move rather than hand-waving.

Type consistency: variantsSeen (service return) is read as priceResult.variantsSeen in Task 4 — consistent. streamManagedVariants(collectionId, onProgress) parameter order is stated in Task 4's Interfaces block and used in that order in Task 4 step 4. onBatchStart(gameId, size) is emitted in Task 3 step 4's flushGame and consumed with both arguments in the batch-composition test. contextFor is defined in Task 2 and called in Task 3.

Price Update Streaming Implementation Plan ​

Global Constraints ​

File Structure ​

Task 1: Golden equivalence harness ​

Task 2: Extract the per-batch body (pure refactor) ​

Task 3: Buffer per game and consume an async iterable ​

Task 4: Stream phase 1 and wire the processor ​

Task 5: Full verification and documentation ​

Self-Review ​

Price Update Streaming Implementation Plan

Global Constraints

File Structure

Task 1: Golden equivalence harness

Task 2: Extract the per-batch body (pure refactor)

Task 3: Buffer per game and consume an async iterable

Task 4: Stream phase 1 and wire the processor

Task 5: Full verification and documentation

Self-Review