← back to Gmc Titlefix
TK-10862: two-phase enumeration (fast Shopify scan + concurrent GMC probe), SCAN_CAP + CONC
95d0cb67e934b2d1bfd825d044027f386e8bd8b8 · 2026-08-27 20:07:49 -0700 · Steve Abrams
Files touched
Diff
commit 95d0cb67e934b2d1bfd825d044027f386e8bd8b8
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Thu Aug 27 20:07:49 2026 -0700
TK-10862: two-phase enumeration (fast Shopify scan + concurrent GMC probe), SCAN_CAP + CONC
---
primary-fix-batch.mjs | 62 +++++++++++++++++++++++++++++++++------------------
1 file changed, 40 insertions(+), 22 deletions(-)
diff --git a/primary-fix-batch.mjs b/primary-fix-batch.mjs
index 64f89e2..02fb0cc 100644
--- a/primary-fix-batch.mjs
+++ b/primary-fix-batch.mjs
@@ -194,16 +194,18 @@ async function main() {
const done = new Set();
if (fs.existsSync(LOG)) for (const ln of fs.readFileSync(LOG, 'utf8').split('\n')) { if (!ln) continue; try { const o = JSON.parse(ln); if (o.action === 'insert' && o.ok) done.add(o.offerId); } catch {} }
- const skip = { notPrimaryDs: 0, alreadyFixed: 0, noImage: 0, badRoll: 0, noSample: 0, processedGetFail: 0, resumeSkip: 0 };
+ const SCAN_CAP = parseInt(process.env.SCAN_CAP || '0', 10); // 0 = unbounded scan (until BATCH_CAP plans or catalog end)
+ const CONC = parseInt(process.env.CONC || '10', 10); // GMC probe concurrency
+ const skip = { notPrimaryDs: 0, alreadyFixed: 0, noImage: 0, badRoll: 0, noSample: 0, processedGet404: 0, processedGetErr: 0, resumeSkip: 0 };
const plans = [];
+ // Phase 1 (fast): Shopify-only scan -> collect leaker candidates (pid, sampleVid, rollVariant, title, handle, vendor).
+ const leakers = [];
let scanned = 0, leaksSeen = 0, pages = 0, cur = null;
- process.stderr.write(`Enumerating active roll-leakers (cap ${BATCH_CAP})...\n`);
+ process.stderr.write(`Phase1: Shopify scan for roll-leakers (SCAN_CAP ${SCAN_CAP || '∞'}, BATCH_CAP ${BATCH_CAP})...\n`);
do {
- if (Date.now() - tokAt > 50 * 60 * 1000) { tok = await token(); tokAt = Date.now(); }
- const d = await gql(`query($c:String){products(first:80,after:$c,query:"status:active"){pageInfo{hasNextPage endCursor} nodes{id handle vendor title onGoogle:publishedOnPublication(publicationId:"${GOOG}") featuredImage{url} variants(first:100){nodes{id sku title price}}}}}`, { c: cur });
+ const d = await gql(`query($c:String){products(first:100,after:$c,query:"status:active"){pageInfo{hasNextPage endCursor} nodes{id handle vendor title onGoogle:publishedOnPublication(publicationId:"${GOOG}") featuredImage{url} variants(first:100){nodes{id sku title price}}}}}`, { c: cur });
const pg = d.products;
for (const p of pg.nodes) {
- if (plans.length >= BATCH_CAP) break;
scanned++;
const vs = p.variants.nodes.map(v => ({ id: vid(v.id), sku: v.sku || '', title: v.title || '', price: parseFloat(v.price) }));
const prices = vs.map(v => v.price).filter(x => !isNaN(x));
@@ -214,33 +216,49 @@ async function main() {
if (!p.featuredImage?.url) { skip.noImage++; continue; }
const rollV = vs.filter(v => v.price > 4.26).sort((a, b) => b.price - a.price)[0];
if (!rollV) { skip.badRoll++; continue; }
- // sample variant = lowest-priced <=4.25 (the memo sample carrying the leaking offer id)
const sampleV = vs.filter(v => v.price <= 4.25).sort((a, b) => a.price - b.price)[0];
if (!sampleV) { skip.noSample++; continue; }
const pid = vid(p.id);
const offerId = `shopify_US_${pid}_${sampleV.id}`;
if (done.has(offerId)) { skip.resumeSkip++; continue; }
- // GET processed sample offer from primary DS
- const { status, body: processed } = await getProcessed(tok, offerId);
- if (status !== 200) { skip.processedGetFail++; continue; }
- if (processed.dataSource !== PRIMARY_DS) { skip.notPrimaryDs++; continue; } // refuse ownership transfer
- // already fixed? served roll price already
- const curPrice = parseFloat(processed.productAttributes?.price?.amountMicros || '0') / 1e6;
- if (curPrice > 4.26) { skip.alreadyFixed++; continue; }
- // capture prestate BEFORE any write
- fs.writeFileSync(ROLLBACK_DIR + offerId + '.json', JSON.stringify({ capturedAt: new Date().toISOString(), dataSource: processed.dataSource, offerId, productAttributes: processed.productAttributes }, null, 2));
- const plan = buildRollBody(processed, rollV, p.title, p.handle);
- if (!plan.productAttributes.imageLink) { skip.noImage++; continue; }
- plans.push({ offerId, vendor: p.vendor || '(none)', handle: p.handle, plan, rollPrice: +rollV.price.toFixed(2), oldPrice: curPrice });
- await sleep(50);
+ leakers.push({ offerId, pid, handle: p.handle, vendor: p.vendor || '(none)', title: p.title, rollV });
}
- pages++; if (pages % 10 === 0) process.stderr.write(` ...scanned ${scanned}, leaksSeen ${leaksSeen}, planned ${plans.length}/${BATCH_CAP}\n`);
+ pages++; if (pages % 5 === 0) process.stderr.write(` ...scanned ${scanned}, leaksSeen ${leaksSeen}, leaker-candidates ${leakers.length}\n`);
cur = pg.pageInfo.hasNextPage ? pg.pageInfo.endCursor : null;
- } while (cur && plans.length < BATCH_CAP);
+ } while (cur && (!SCAN_CAP || scanned < SCAN_CAP));
+ process.stderr.write(`Phase1 done: scanned ${scanned}, leaker-candidates ${leakers.length}. Phase2: GMC probe (conc ${CONC})...\n`);
+
+ // Phase 2 (concurrent): probe each leaker's sample offer in primary DS; keep only PRIMARY-DS + still-$4.25.
+ let idx = 0, probed = 0;
+ async function worker() {
+ while (idx < leakers.length && plans.length < BATCH_CAP) {
+ const lk = leakers[idx++];
+ if (Date.now() - tokAt > 50 * 60 * 1000) { tok = await token(); tokAt = Date.now(); }
+ const { status, body: processed } = await getProcessed(tok, lk.offerId);
+ probed++;
+ if (status === 404) { skip.processedGet404++; }
+ else if (status !== 200) { skip.processedGetErr++; }
+ else if (processed.dataSource !== PRIMARY_DS) { skip.notPrimaryDs++; }
+ else {
+ const curPrice = parseFloat(processed.productAttributes?.price?.amountMicros || '0') / 1e6;
+ if (curPrice > 4.26) { skip.alreadyFixed++; }
+ else {
+ const plan = buildRollBody(processed, lk.rollV, lk.title, lk.handle);
+ if (!plan.productAttributes.imageLink) { skip.noImage++; }
+ else if (plans.length < BATCH_CAP) {
+ fs.writeFileSync(ROLLBACK_DIR + lk.offerId + '.json', JSON.stringify({ capturedAt: new Date().toISOString(), dataSource: processed.dataSource, offerId: lk.offerId, productAttributes: processed.productAttributes }, null, 2));
+ plans.push({ offerId: lk.offerId, vendor: lk.vendor, handle: lk.handle, plan, rollPrice: +lk.rollV.price.toFixed(2), oldPrice: curPrice });
+ }
+ }
+ }
+ if (probed % 200 === 0) process.stderr.write(` ...probed ${probed}/${leakers.length}, eligible-plans ${plans.length}/${BATCH_CAP}\n`);
+ }
+ }
+ await Promise.all(Array.from({ length: CONC }, worker));
const summary = {
generated_at: new Date().toISOString(), ticket: 'TK-10862', mechanism: 'TK-10451 primary-fix (productInputs:insert -> primary DS 180695450, sample-offerId supersede)',
- mode: APPLY ? 'APPLY' : 'DRY-RUN', cap: BATCH_CAP, scanned_active: scanned, leaks_seen: leaksSeen, planned: plans.length, skipped: skip,
+ mode: APPLY ? 'APPLY' : 'DRY-RUN', cap: BATCH_CAP, scan_cap: SCAN_CAP || null, conc: CONC, scanned_active: scanned, leaks_seen: leaksSeen, leaker_candidates: leakers.length, planned: plans.length, skipped: skip,
roll_price_dist: (() => { const a = plans.map(x => x.rollPrice).sort((x, y) => x - y); return a.length ? { min: a[0], median: a[a.length >> 1], max: a[a.length - 1] } : {}; })(),
vendors: [...new Set(plans.map(x => x.vendor))].length, cost: '$0 (Content API free)'
};
← b76a6a6 TK-10862: scaled primary-fix batch runner (capped, reversibl
·
back to Gmc Titlefix
·
auto-data-snapshot: 2026-08-27T20:01:28 (250 data files) — d 3b8cf76 →