Почему не работает COMMIT в Postgres?

Question

Прошу помочь разобраться, почему не работает COMMIT в Postgres?

Решаю такую задачу. Есть API с постраничной выдачей. Делаю синхронные постраничные запросы по 1000 объектов. Эти объекты асинхронно вставляю в таблицу БД. У меня есть код на Node.js + pg module, который делает это. Скрипт работает с БД под отдельным пользователем. Однако мне непонятно следующее поведение: запускаю скрипт и параллельно через другого пользователя в pgAdmin запросом select count(*) from objects смотрю число объектов, загруженных в таблицу. Оно не растёт, не смотря на то, что в скрипте явно открываю транзакцию и делаю коммиты по её завершению. Мне удаётся увидеть изменения только лишь после того, как скрипт полностью завершил работу. Если прервать выполнение скрипта вручную через ctrl-c, то коммиты, которые уже были сделаны тоже не сохраняются. Хотелось бы понять, почему так происходит, пробовал разные варианты, смотрел логи Postgres, коммиты там делаются, но изменений не видно, пока скрипт не отработает полностью.

Далее код: файл db.js

const Pool = require('pg').Pool
module.exports = Pool

файл index.js

const Pool = require('./db')
const itemsCount = 1000
const apiList = [
    'https://api1.example.com'
    'https://api2.example.com'
    'https://api3.example.com'
]
const dbOptions = {
    host: 'localhost',
    port: 5432,
    database: 'db_name',
    user: 'user',
    password: 'password',
    max: 20
}
apiList.forEach(api => {
    const url = api
    let page = 1
    let objects = []
    do {
        objects = getObjects(url, page) // эта функция постранично получает данные из API по 1000 штук.
        processObjects(objects, api) // эта функция загружает полученные объекты в БД.
        page++
    } while (objects.length === itemsCount)

})
async function processObjects(objects, api) {
    const pool = new Pool(dbOptions)
    for (const object of objects) {
        const ID = object.ID
        const client = await pool.connect()
        const selectObjects = await client.query('select * from objects where id=$1', [ID])
        if (selectObjects.rows.length === 0) {
            try {
                await client.query('BEGIN;')
                const insertObjects = await client.query(insert into objects ("id", "json", "api", "loaded") VALUES ($1, $2, $3, $4), [ID, object, api, 'now()'])
                await client.query('COMMIT;')
            } catch (error) {
                console.error(error.stack)
            }
        }
        await client.end()
        client.release()
    }
    await pool.end()
}

Answer 1

Вот так нужно написать (getObjects - заглушка для примера):

// @ts-check
const { setTimeout } = require('node:timers/promises');
const { randomUUID } = require('node:crypto');
const { Pool } = require('pg');
const pool = new Pool({
    host: 'localhost',
    port: 5432,
    database: 'db_name',
    user: 'user',
    password: 'password',
    max: 20
});
run().catch(console.error);
async function run() {
  const itemsCount = 1000;
const apiList = [
    'https://api1.example.com',
    'https://api2.example.com',
    'https://api3.example.com',
  ];
const res = await Promise.allSettled(
    apiList.map(async (url) => {
      console.log([START] ${url});
      await handleApiItem(url, itemsCount);
      console.log([DONE] ${url});
    }),
  );
  res.forEach((x) => {
    if (x.status === 'rejected') {
      console.error(x.reason);
    }
  });
await pool.end().catch(console.error);
}
/**

Фyнкция, которая загрузит все данные из endpoint
@param {string} url
@param {number} [requestedItemsCount]

*/
async function handleApiItem(url, requestedItemsCount = 1000) {
  let page = 1;
  while (1) {
    try {
      const objects = await getObjects(
        url,
        { pageNumber: page, itemsCount: requestedItemsCount }
      );
      if (objects?.length > 0) {
        await processObjects(objects, url);
      }
      if (!objects || objects.length < requestedItemsCount) {
        break;
      }
      page++;

    } catch (err) {
      console.error(err);
    }
  }
}
/**

Эта функция загружает полученные объекты в БД.
@param {Awaited<ReturnType<getObjects>>} objects
@param {string} api

*/
async function processObjects(objects, api) {
  const client = await pool.connect();
for (const object of objects) {
    const ID = object.ID;
    const selectObjects = await client.query('select id from objects where id=$1', [ID]);
    if (selectObjects.rows.length === 0) {
      try {
        await client.query('BEGIN;');
        await client.query(
          insert into objects (&quot;id&quot;, &quot;json&quot;, &quot;api&quot;, &quot;loaded&quot;) VALUES ($1, $2, $3, $4),
          [ID, object, api, 'now()']
        );
        await client.query('COMMIT;')
      } catch (error) {
        console.error(error);
      }
    }
  }
client.release();
}
/**

Эта функция постранично получает данные из API по 1000 штук.
@param {string} url
@param {{ pageNumber: number, itemsCount?: number }} options

/
async function getObjects(url, options) {
  const timeout = Math.trunc(10  Math.random() + 1);
  await setTimeout(timeout);
return options.pageNumber > 10 && timeout % 5 === 0
    ? []
    : Array.from(
        { length: options.itemsCount || 1000 },
        () => ({ ID: randomUUID() })
      );
}

я бы дал совет в const selectObjects = await client.query('select id from objects where id=$1', [ID]); добавить and api=$2.