fix(backfill): merge-preserving regen, real recipe yields, single-flight resolve

tools/price-backfill/backfill.py

@@ -46,6 +46,7 @@ def __init__(self, build_id, sentinel_id=4389):
         self.generation = 0
         self.consecutive_404 = 0
         self.sentinel_id = sentinel_id  # known-good item; confirms a real redeploy
+        self._resolving = False  # single-flight guard for the network calls below
 
     def snapshot(self):
         with self._lock:
@@ -65,20 +66,38 @@ def note_result(self, was_404, threshold):
             # item that always has data. This stops no-data streaks from triggering
             # a needless buildId re-resolve storm and recovery re-fetch.
             self.consecutive_404 = 0
+            if self._resolving:
+                # Another thread is already running the sentinel/redeploy check;
+                # don't pile on redundant network calls, just proceed on the
+                # buildId as-is, same as a thread that hasn't hit the threshold.
+                return self.build_id
+            self._resolving = True
+            build_id = self.build_id
+
+        # Network I/O happens outside the lock. Only the thread that won the
+        # _resolving flag above reaches here, so this stays single-flight.
+        try:
             try:
-                sentinel = wowauctions.fetch_item(self.build_id, self.sentinel_id)
+                sentinel = wowauctions.fetch_item(build_id, self.sentinel_id)
             except Exception:
-                return self.build_id  # transient error; do not assume redeploy
+                return build_id  # transient error; do not assume redeploy
             if sentinel is not None:
-                return self.build_id  # sentinel still resolves -> buildId is fine
+                return build_id  # sentinel still resolves -> buildId is fine
             # Sentinel 404s under the current buildId -> real redeploy. Re-resolve,
             # and bump the generation only if the buildId actually changed, so the
             # recovery pass stays scoped to items fetched under the stale buildId.
             new_build_id = wowauctions.resolve_build_id()
-            if new_build_id != self.build_id:
-                self.build_id = new_build_id
-                self.generation += 1
-            return self.build_id
+            with self._lock:
+                if new_build_id != self.build_id:
+                    self.build_id = new_build_id
+                    self.generation += 1
+                return self.build_id
+        finally:
+            # Released only after the buildId update above; a thread crossing
+            # the threshold mid-resolve must not start a redundant resolve
+            # against the still-stale buildId.
+            with self._lock:
+                self._resolving = False
 
 
 def process_id(item_id, state, cfg, existing, now, rate_delay, redeploy_threshold, ckpt_fh, ckpt_lock):
@@ -120,8 +139,16 @@ def process_id(item_id, state, cfg, existing, now, rate_delay, redeploy_threshol
     return rec
 
 
-def write_outputs(records, out_sql, skipped_csv, deviations_csv, item_counts_csv=None):
-    rows = [tuple(r["row"]) for r in records if r.get("row")]
+def write_outputs(records, existing, out_sql, skipped_csv, deviations_csv, item_counts_csv=None):
+    # --out-sql defaults to --existing-sql (in-place regeneration): seed from
+    # existing overrides, then update/add only items present in this run's
+    # records, so items absent from candidates.csv aren't silently dropped.
+    merged = dict(existing)
+    for r in records:
+        if r.get("row"):
+            item, avg, mn = r["row"]
+            merged[item] = (avg, mn)
+    rows = [(item, avg, mn) for item, (avg, mn) in merged.items()]
     sqlio.write_override_sql(out_sql, rows)
 
     with open(skipped_csv, "w", newline="", encoding="utf-8") as fh:
@@ -243,8 +270,8 @@ def run_pass(id_list):
             run_pass(failed)
 
     records = list(load_checkpoint(args.checkpoint).values())
-    kept, ndev = write_outputs(records, out_sql, args.skipped_csv, args.deviations_csv,
-                               args.item_counts_csv)
+    kept, ndev = write_outputs(records, existing, out_sql, args.skipped_csv,
+                               args.deviations_csv, args.item_counts_csv)
     print("wrote {} rows to {} | {} deviations | build_gen={}".format(
         kept, out_sql, ndev, state.generation))
 

tools/price-backfill/spelldbc.py

@@ -1,10 +1,12 @@
 """Parse WotLK 3.3.5a Spell.dbc into a created-item -> recipe map."""
 import struct
+import sys
 from typing import Dict, List, Tuple
 
 CREATE_ITEM_EFFECTS = (24, 66)
 _ID = 0
 _EFFECT = (71, 72, 73)
+_EFFECT_BASE_POINTS = (80, 81, 82)
 _EFFECT_ITEM = (107, 108, 109)
 _REAGENT = range(52, 60)
 _REAGENT_COUNT = range(60, 68)
@@ -30,11 +32,19 @@ def parse_spell_dbc(path: str) -> Dict[int, Tuple[int, List[Tuple[int, int]]]]:
                 reagents.append((item, cnt))
         if not reagents:
             continue
-        for ei, ii in zip(_EFFECT, _EFFECT_ITEM):
+        for ei, ii, bi in zip(_EFFECT, _EFFECT_ITEM, _EFFECT_BASE_POINTS):
             if fields[ei] in CREATE_ITEM_EFFECTS and fields[ii] > 0:
                 out = fields[ii]
+                # 3.3.5a yield is EffectBasePoints + 1 plus a random roll; the
+                # roll isn't determinable from static DBC data, so basePoints + 1
+                # is the minimum yield either way. basePoints is signed -- a
+                # non-positive yield means this isn't a real crafting yield.
+                yield_amt = fields[bi] + 1
+                if yield_amt <= 0:
+                    print("spelldbc: skipping item %d: non-positive yield %d" % (fields[ii], yield_amt), file=sys.stderr)
+                    continue
                 # first recipe with reagents wins; keep deterministic
-                recipes.setdefault(out, (1, reagents))
+                recipes.setdefault(out, (yield_amt, reagents))
     return recipes
 
 

tools/price-backfill/test_backfill.py

@@ -3,10 +3,12 @@
 import os
 import tempfile
 import threading
+import time
 import unittest
 from datetime import datetime
 
 import backfill
+import sqlio
 import transform
 import wowauctions
 
@@ -172,5 +174,102 @@ def test_non404_resets_counter(self):
         self.assertEqual(state.consecutive_404, 0)
 
 
+class WriteOutputsMergeTest(unittest.TestCase):
+    """--out-sql defaults to --existing-sql (in-place regeneration): an item
+    already overridden must survive even if it's absent from this run's
+    candidates, instead of silently reverting to default AH pricing."""
+
+    def setUp(self):
+        fd, self.out_sql = tempfile.mkstemp(suffix=".sql")
+        os.close(fd)
+        fd, self.skipped = tempfile.mkstemp(suffix=".csv")
+        os.close(fd)
+        fd, self.deviations = tempfile.mkstemp(suffix=".csv")
+        os.close(fd)
+
+    def tearDown(self):
+        for path in (self.out_sql, self.skipped, self.deviations):
+            if os.path.exists(path):
+                os.remove(path)
+
+    def test_existing_item_absent_from_candidates_survives(self):
+        existing = {99: (500, 400), 1: (10, 5)}
+        records = [{"item": 1, "row": [1, 20, 15], "reason": None}]
+        backfill.write_outputs(records, existing, self.out_sql, self.skipped,
+                                self.deviations)
+        result = sqlio.parse_override_sql(self.out_sql)
+        # item 99 was not part of this run's candidates but is a pre-existing
+        # override -- it must be carried through unchanged.
+        self.assertEqual(result[99], (500, 400))
+        # item 1 was re-derived this run -- the new values win.
+        self.assertEqual(result[1], (20, 15))
+
+
+class NoteResultLockScopeTest(unittest.TestCase):
+    """The lock must protect shared-state reads/writes only, not the
+    sentinel-fetch / resolve_build_id network calls, so a redeploy-sentinel
+    check on one thread doesn't collapse concurrency to 1 for everyone else."""
+
+    def setUp(self):
+        self._orig_fetch = wowauctions.fetch_item
+        self._orig_resolve = wowauctions.resolve_build_id
+
+    def tearDown(self):
+        wowauctions.fetch_item = self._orig_fetch
+        wowauctions.resolve_build_id = self._orig_resolve
+
+    def test_lock_not_held_during_network_call(self):
+        entered = threading.Event()
+        release = threading.Event()
+
+        def slow_fetch(build_id, item_id):
+            entered.set()
+            release.wait(2)
+            return None  # sentinel 404s -> proceeds to resolve_build_id
+
+        wowauctions.fetch_item = slow_fetch
+        wowauctions.resolve_build_id = lambda: "OLD"  # buildId unchanged
+
+        state = backfill.BuildIdState("OLD", sentinel_id=4389)
+        t = threading.Thread(target=lambda: state.note_result(True, 1))
+        t.start()
+        self.assertTrue(entered.wait(2), "sentinel fetch never started")
+        # The sentinel fetch is in flight; the lock must be free so other
+        # threads (e.g. via snapshot()) are not blocked for the network RTT.
+        acquired = state._lock.acquire(timeout=1)
+        self.assertTrue(acquired, "lock is still held during network I/O")
+        state._lock.release()
+        release.set()
+        t.join(2)
+
+    def test_only_one_thread_performs_reresolve(self):
+        resolve_calls = []
+        call_lock = threading.Lock()
+
+        def counting_resolve():
+            with call_lock:
+                resolve_calls.append(1)
+            time.sleep(0.05)
+            return "NEW"
+
+        wowauctions.fetch_item = lambda build_id, item_id: None  # always 404
+        wowauctions.resolve_build_id = counting_resolve
+
+        state = backfill.BuildIdState("OLD", sentinel_id=4389)
+        threads = [threading.Thread(target=lambda: state.note_result(True, 1))
+                   for _ in range(8)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(2)
+
+        # Even though every thread crossed the threshold concurrently, only
+        # one should have actually re-resolved; the rest proceed on the
+        # (old, then updated) buildId per the single-flight design.
+        self.assertEqual(len(resolve_calls), 1)
+        self.assertEqual(state.build_id, "NEW")
+        self.assertEqual(state.generation, 1)
+
+
 if __name__ == "__main__":
     unittest.main()

tools/price-backfill/test_spelldbc.py

@@ -3,6 +3,7 @@
 import os
 import unittest
 
+import recipe_prices
 import spelldbc
 
 FIELD_COUNT = 234
@@ -56,5 +57,55 @@ def test_validate_ok_and_fail(self):
         big[4389] = (1, [(3575, 1), (10558, 1)])
         spelldbc.validate(big)  # no raise
 
+
+class YieldExtractionTest(unittest.TestCase):
+    """EffectBasePoints (field 80-82) drives the recipe yield instead of a
+    hardcoded 1: yield = basePoints + 1 (WotLK 3.3.5a effect-value formula)."""
+
+    def _write(self, records):
+        fd, path = tempfile.mkstemp(suffix=".dbc")
+        os.write(fd, _dbc(records)); os.close(fd)
+        self.addCleanup(os.remove, path)
+        return path
+
+    def test_yield_one_recipe_unchanged(self):
+        # basePoints unset (0) -> yield 1, same as before this fix.
+        rec = {0: 4001, 71: 24, 107: 5001, 52: 3575, 60: 1}
+        path = self._write([_record(rec)])
+        recipes = spelldbc.parse_spell_dbc(path)
+        yld, reagents = recipes[5001]
+        self.assertEqual(yld, 1)
+        self.assertEqual(reagents, [(3575, 1)])
+
+    def test_fixed_multi_yield_recipe(self):
+        # basePoints=4, dieSides unset (0, non-ambiguous) -> yield 5.
+        rec = {0: 4002, 71: 24, 107: 5002, 52: 3575, 60: 2, 80: 4}
+        path = self._write([_record(rec)])
+        recipes = spelldbc.parse_spell_dbc(path)
+        yld, reagents = recipes[5002]
+        self.assertEqual(yld, 5)
+        self.assertEqual(reagents, [(3575, 2)])
+
+    def test_multi_yield_propagates_to_recursive_recipe_cost(self):
+        # Tier-1: item 5002 crafted from 2x reagent 3575 (cc-priced 100), yield 5
+        # -> per-unit cost = 200 / 5 = 40.
+        rec = {0: 4002, 71: 24, 107: 5002, 52: 3575, 60: 2, 80: 4}
+        path = self._write([_record(rec)])
+        recipes = dict(spelldbc.parse_spell_dbc(path))
+        # Tier-2: item 6000 crafted from 1x item 5002 (the tier-1 output).
+        recipes[6000] = (1, [(5002, 1)])
+        cfg = recipe_prices.DeriveConfig(margin_avg=1.0, margin_min=1.0, max_depth=10)
+        cost = recipe_prices.resolve_price(6000, recipes, {3575: 100}, {}, cfg)
+        self.assertEqual(cost, 40)
+
+    def test_negative_base_points_recipe_skipped(self):
+        # Negative basePoints (yield <= 0) means this isn't a real crafting
+        # yield -> the recipe is skipped rather than priced as yield 1.
+        rec = {0: 4003, 71: 24, 107: 5003, 52: 3575, 60: 1, 80: -5}
+        path = self._write([_record(rec)])
+        recipes = spelldbc.parse_spell_dbc(path)
+        self.assertNotIn(5003, recipes)
+
+
 if __name__ == "__main__":
     unittest.main()