S5a-es: draw Elm purchases of batch offers from the live batch (fail-soft)

.github/workflows/test.yml

@@ -0,0 +1,31 @@
+name: Test
+
+on:
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:17
+        env:
+          POSTGRES_HOST_AUTH_METHOD: trust
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          # The box's version (see .nvmrc); upgrading it is separate work.
+          node-version-file: .nvmrc
+      - name: Install libpq headers (pg-native)
+        run: sudo apt-get update && sudo apt-get install -y libpq-dev
+      - run: yarn install --frozen-lockfile
+      - run: yarn format
+      - run: yarn test

.nvmrc

@@ -0,0 +1 @@
+16.20.2

package.json

@@ -5,8 +5,8 @@
   "main": "src/app.js",
   "scripts": {
     "start": "standard ./src/ && node src/app.js",
-    "format": "standard ./src/",
-    "test": "echo \"Error: no test specified\" && exit 1"
+    "format": "standard --env mocha ./src/ ./test/",
+    "test": "mocha --exit test/*.test.js"
   },
   "author": "Julien Lucca",
   "license": "MIT",
@@ -19,6 +19,7 @@
     "pg-native": "^3.0.0"
   },
   "devDependencies": {
+    "mocha": "^10.8.2",
     "standard": "^12.0.1"
   }
 }

src/app.js

@@ -6,6 +6,7 @@ const { GetActionsHandler } = require('./GetActionsHandler')
 const { logInit, logExit } = require('./logging')
 
 const updaters = require('./updaters')
+const { checkDrawFunction } = require('./updaters/community')
 const effects = []
 
 const http = require('http')
@@ -42,6 +43,10 @@ async function init () {
        )`
     )
 
+    // S5: log 'S5 draw function missing' if the backend's draw function or
+    // order_items.draw_outcome isn't there; indexing carries on either way.
+    await checkDrawFunction(db)
+
     const actionHandler = new GetActionsHandler(
       updaters,
       effects,

src/updaters/community.js

@@ -7,6 +7,70 @@ const config = require(`../config/${process.env.NODE_ENV || 'dev'}`)
 const { resolveClaimId, resolveCreatedActionId, resolveCreatedObjectiveId, ResolveError } = require('../chain')
 const { toUTC } = require('../dates')
 
+// S5a-es (plan §9): an Elm purchase of a batch offer takes its units from the live batch
+// through the backend's frozen DB function. Transient database errors are retried; anything
+// else fails soft (order recorded unassigned) so a draw problem never halts the indexer.
+const DRAW_TRANSIENT_CODES = ['40P01', '40001', '55P03']
+const DRAW_MAX_RETRIES = 3
+const DRAW_RETRY_DELAY_MS = 50
+
+const sleep = ms => new Promise(resolve => setTimeout(resolve, ms))
+
+// Startup check, re-run after every draw error: the draw function and the
+// order_items.draw_outcome column must exist (massive reads the schema once at startup and
+// rejects unknown columns). The hourly cron greps the log for 'S5 draw function missing';
+// indexing carries on regardless.
+async function checkDrawFunction (db) {
+  try {
+    const row = await db.instance.one(
+      `SELECT to_regprocedure('public.shop_draw_from_batch(bigint,integer,timestamp)') IS NOT NULL AS has_function,
+              EXISTS (SELECT 1 FROM information_schema.columns
+                       WHERE table_schema = 'public' AND table_name = 'order_items'
+                         AND column_name = 'draw_outcome') AS has_column`
+    )
+    if (!row.has_function || !row.has_column) {
+      console.error(`S5 draw function missing (function: ${row.has_function}, order_items.draw_outcome: ${row.has_column})`)
+      return false
+    }
+    return true
+  } catch (e) {
+    logError('S5 draw function check failed', e)
+    return false
+  }
+}
+
+// Draw `units` from the batch live at `at` (the block's time, never the wall clock).
+// Each attempt is a savepoint (a nested pg-promise transaction), so a failed draw rolls
+// back only itself and the outer transaction carries on.
+// Resolves { batchId, outcome } with outcome 'drawn' | 'no_fit' | 'error'.
+async function drawFromBatch (tx, productId, units, at) {
+  for (let attempt = 0; ; attempt++) {
+    try {
+      const row = await tx.instance.tx(async sp => {
+        const { previous } = await sp.one(`SELECT current_setting('lock_timeout') AS previous`)
+        await sp.none(`SET LOCAL lock_timeout = '2s'`)
+        const drawn = await sp.oneOrNone(
+          'SELECT public.shop_draw_from_batch($1, $2, $3) AS batch_id',
+          [productId, units, at]
+        )
+        // SET LOCAL outlives a released savepoint; put the caller's timeout back.
+        await sp.one(`SELECT set_config('lock_timeout', $1, true)`, [previous])
+        return drawn
+      })
+      const batchId = row && row.batch_id != null ? row.batch_id : null
+      return { batchId, outcome: batchId === null ? 'no_fit' : 'drawn' }
+    } catch (e) {
+      if (DRAW_TRANSIENT_CODES.includes(e.code) && attempt < DRAW_MAX_RETRIES) {
+        await sleep(DRAW_RETRY_DELAY_MS)
+        continue
+      }
+      logError('S5 draw failed; order recorded unassigned', e)
+      await checkDrawFunction(tx)
+      return { batchId: null, outcome: 'error' }
+    }
+  }
+}
+
 async function createCommunity (db, payload, blockInfo) {
   console.log(`Cambiatus >>> Create Community`, blockInfo.blockNumber)
 
@@ -245,17 +309,25 @@ async function transferSale (db, payload, blockInfo, context) {
     })
     if (Number(existing) > 0) return
 
-    const whereArg = {
-      id: payload.data.sale_id,
-      is_deleted: false
-    }
+    const whereArg = { id: payload.data.sale_id }
 
-    // Find sale
+    // Find sale. Deleted offers are still found: a purchase signed just as the seller
+    // deleted the offer is recorded like any other. A product that truly doesn't exist
+    // is a ResolveError (skipped, paged, re-indexable) rather than a crash loop.
     const sale = await tx.products.findOne(whereArg)
     if (sale === null) {
-      throw new Error('No sale data available')
+      throw new ResolveError(`transfersale: product ${payload.data.sale_id} not found`)
     }
 
+    const quantity = parseInt(payload.data.units) || 1
+
+    // Batch offers draw from the live batch at the block's time (a batch offer has
+    // track_stock false, so the units decrement below is skipped). The replay guard above
+    // already ran, so re-indexing a block never draws twice.
+    const draw = sale.availability === 'batches'
+      ? await drawFromBatch(tx, sale.id, quantity, toUTC(blockInfo.timestamp))
+      : null
+
     if (sale.track_stock) {
       const newUnits = sale.units - parseInt(payload.data.units)
 
@@ -290,16 +362,21 @@ async function transferSale (db, payload, blockInfo, context) {
 
     // Write the matching line item. Kept inside the same guarded path so a
     // replay that hits orders_dedup_idx above never reaches this insert.
-    const quantity = parseInt(payload.data.units) || 1
-    await tx.order_items.insert({
+    const item = {
       order_id: order.id,
       product_id: sale.id,
       quantity: quantity,
       unit_price_token: amount / Math.max(quantity, 1),
       title_snapshot: sale.title,
       inserted_at: toUTC(blockInfo.timestamp),
       updated_at: toUTC(blockInfo.timestamp)
-    })
+    }
+    // Only batch offers carry the new keys; every other insert is exactly as before.
+    if (draw !== null) {
+      item.batch_id = draw.batchId
+      item.draw_outcome = draw.outcome
+    }
+    await tx.order_items.insert(item)
   }
 
   // Return the transaction so ledgered() can await it. Without the return the updater
@@ -936,6 +1013,7 @@ async function assignRole (db, payload, blockInfo, _context) {
 }
 
 module.exports = {
+  checkDrawFunction,
   createCommunity,
   updateCommunity,
   netlink,

test/fixtures/generate_s5_schema.js

@@ -0,0 +1,82 @@
+// Generates test/fixtures/s5_schema.sql: the tables transferSale writes, built from
+// written_columns.json (the manifest shared with the backend repo), plus the fixed
+// definitions copied from the backend's S5a-core migration (20261017100000_s5a_batches.exs).
+// Run `node test/fixtures/generate_s5_schema.js > test/fixtures/s5_schema.sql` after the
+// manifest changes; the test suite fails if the committed file is stale.
+const manifest = require('./written_columns.json')
+
+// Not in the manifest (the indexer only UPDATEs products.units and reads the rest), so
+// declared by hand: the columns transferSale reads.
+const PRODUCTS = `CREATE TABLE products (
+  id bigserial PRIMARY KEY,
+  title character varying(255),
+  units integer,
+  is_deleted boolean DEFAULT false,
+  track_stock boolean NOT NULL DEFAULT true,
+  availability character varying(255)
+);`
+
+// Copied from the backend migration; the draw function is a frozen API.
+const FIXED = `CREATE EXTENSION IF NOT EXISTS btree_gist;
+
+CREATE TABLE product_batches (
+  id bigserial PRIMARY KEY,
+  product_id bigint NOT NULL REFERENCES products(id),
+  quantity integer NOT NULL,
+  remaining integer NOT NULL,
+  label character varying(255),
+  opens_at timestamp(0) without time zone NOT NULL,
+  closes_at timestamp(0) without time zone NOT NULL,
+  inserted_at timestamp(0) without time zone NOT NULL DEFAULT now(),
+  updated_at timestamp(0) without time zone NOT NULL DEFAULT now(),
+  delivery_on date,
+  closed_at timestamp(0) without time zone,
+  CONSTRAINT product_batches_quantity_range CHECK (quantity BETWEEN 1 AND 200),
+  CONSTRAINT product_batches_remaining_non_negative CHECK (remaining >= 0),
+  CONSTRAINT product_batches_remaining_lte_quantity CHECK (remaining <= quantity),
+  CONSTRAINT product_batches_window_order CHECK (closes_at > opens_at)
+);
+
+ALTER TABLE product_batches ADD CONSTRAINT product_batches_no_overlap
+  EXCLUDE USING gist (product_id WITH =, tsrange(opens_at, closes_at, '[)') WITH &&)
+  WHERE (closed_at IS NULL);
+
+CREATE TABLE _processed_actions (
+  global_seq bigint PRIMARY KEY,
+  processed_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE FUNCTION public.shop_draw_from_batch(p_product_id bigint, p_quantity integer, p_at timestamp)
+RETURNS bigint LANGUAGE plpgsql AS $$
+DECLARE v_batch_id bigint;
+BEGIN
+  PERFORM 1 FROM public.products WHERE id = p_product_id FOR NO KEY UPDATE;
+  UPDATE public.product_batches SET remaining = remaining - p_quantity
+   WHERE product_id = p_product_id AND closed_at IS NULL
+     AND opens_at <= p_at AND closes_at > p_at
+     AND remaining >= p_quantity AND p_quantity > 0
+  RETURNING id INTO v_batch_id;
+  RETURN v_batch_id;
+END $$;`
+
+function table (name, spec) {
+  const columns = Object.entries(spec.columns).map(([column, type]) => `  ${column} ${type}`)
+  return `CREATE TABLE ${name} (\n  id bigserial PRIMARY KEY,\n${columns.join(',\n')}\n);`
+}
+
+function generate () {
+  return [
+    '-- GENERATED by test/fixtures/generate_s5_schema.js from written_columns.json. Do not edit.',
+    PRODUCTS,
+    FIXED,
+    table('orders', manifest.orders),
+    'CREATE UNIQUE INDEX orders_dedup_idx ON orders (created_tx, from_id, product_id);',
+    table('order_items', manifest.order_items),
+    'ALTER TABLE order_items ADD CONSTRAINT order_items_draw_outcome_check CHECK (draw_outcome IN (\'drawn\', \'no_fit\', \'error\'));',
+    ''
+  ].join('\n\n')
+}
+
+if (require.main === module) process.stdout.write(generate())
+
+module.exports = { generate }

test/fixtures/s5_schema.sql

@@ -0,0 +1,89 @@
+-- GENERATED by test/fixtures/generate_s5_schema.js from written_columns.json. Do not edit.
+
+CREATE TABLE products (
+  id bigserial PRIMARY KEY,
+  title character varying(255),
+  units integer,
+  is_deleted boolean DEFAULT false,
+  track_stock boolean NOT NULL DEFAULT true,
+  availability character varying(255)
+);
+
+CREATE EXTENSION IF NOT EXISTS btree_gist;
+
+CREATE TABLE product_batches (
+  id bigserial PRIMARY KEY,
+  product_id bigint NOT NULL REFERENCES products(id),
+  quantity integer NOT NULL,
+  remaining integer NOT NULL,
+  label character varying(255),
+  opens_at timestamp(0) without time zone NOT NULL,
+  closes_at timestamp(0) without time zone NOT NULL,
+  inserted_at timestamp(0) without time zone NOT NULL DEFAULT now(),
+  updated_at timestamp(0) without time zone NOT NULL DEFAULT now(),
+  delivery_on date,
+  closed_at timestamp(0) without time zone,
+  CONSTRAINT product_batches_quantity_range CHECK (quantity BETWEEN 1 AND 200),
+  CONSTRAINT product_batches_remaining_non_negative CHECK (remaining >= 0),
+  CONSTRAINT product_batches_remaining_lte_quantity CHECK (remaining <= quantity),
+  CONSTRAINT product_batches_window_order CHECK (closes_at > opens_at)
+);
+
+ALTER TABLE product_batches ADD CONSTRAINT product_batches_no_overlap
+  EXCLUDE USING gist (product_id WITH =, tsrange(opens_at, closes_at, '[)') WITH &&)
+  WHERE (closed_at IS NULL);
+
+CREATE TABLE _processed_actions (
+  global_seq bigint PRIMARY KEY,
+  processed_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE FUNCTION public.shop_draw_from_batch(p_product_id bigint, p_quantity integer, p_at timestamp)
+RETURNS bigint LANGUAGE plpgsql AS $$
+DECLARE v_batch_id bigint;
+BEGIN
+  PERFORM 1 FROM public.products WHERE id = p_product_id FOR NO KEY UPDATE;
+  UPDATE public.product_batches SET remaining = remaining - p_quantity
+   WHERE product_id = p_product_id AND closed_at IS NULL
+     AND opens_at <= p_at AND closes_at > p_at
+     AND remaining >= p_quantity AND p_quantity > 0
+  RETURNING id INTO v_batch_id;
+  RETURN v_batch_id;
+END $$;
+
+CREATE TABLE orders (
+  id bigserial PRIMARY KEY,
+  product_id bigint,
+  from_id character varying,
+  to_id character varying,
+  amount double precision,
+  units integer,
+  community_id character varying,
+  status character varying,
+  origin character varying,
+  total_token numeric,
+  created_block integer,
+  created_tx character varying,
+  created_at timestamp without time zone,
+  created_eos_account character varying,
+  inserted_at timestamp without time zone,
+  updated_at timestamp without time zone
+);
+
+CREATE UNIQUE INDEX orders_dedup_idx ON orders (created_tx, from_id, product_id);
+
+CREATE TABLE order_items (
+  id bigserial PRIMARY KEY,
+  order_id bigint,
+  product_id bigint,
+  batch_id bigint,
+  quantity integer,
+  unit_price_token numeric,
+  title_snapshot character varying,
+  draw_outcome character varying,
+  inserted_at timestamp without time zone,
+  updated_at timestamp without time zone
+);
+
+ALTER TABLE order_items ADD CONSTRAINT order_items_draw_outcome_check CHECK (draw_outcome IN ('drawn', 'no_fit', 'error'));
+