From 53a0d7ded324164ab515eb206689e99d8dbec4b9 Mon Sep 17 00:00:00 2001 From: Vitor Date: Fri, 6 Nov 2026 14:22:45 -0300 Subject: [PATCH 1/2] fix(apps): Evita pedidos duplicados no Tiny e no Bling ERP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cada pedido gera dois eventos `orders-anyStatusSet` em segundos (checkout e pagamento), e nada impede que as duas execuções rodem ao mesmo tempo: `maxInstances` é best-effort e é estourado em cold starts simultâneos. A única proteção contra criar o pedido duas vezes no ERP era consultá-lo antes, o que não basta porque a criação leva segundos e as duas execuções leem "não existe" antes de qualquer uma gravar. Em 06/11/2026 isso duplicou 7 de 31 pedidos de uma loja em produção (zero nos 30 dias anteriores, em 1218 pedidos), quando o GCF passou a subir duas instâncias no mesmo instante. Adiciona `tryLockExport`/`unlockExport` em `@cloudcommerce/firebase`: um `create()` no Firestore, que falha com ALREADY_EXISTS para a segunda execução. Quem perde o lock espera e procura o pedido de novo, seguindo para a atualização de status em vez de criar. O lock é liberado quando o POST falha, e é assumido após 5 min para não travar o pedido se a execução vencedora morrer antes de liberá-lo. Co-Authored-By: Claude Opus 5 --- .../src/integration/export-order-to-bling.ts | 25 +++++ .../src/integration/export-order-to-tiny.ts | 98 ++++++++++++------- packages/firebase/package.json | 1 + packages/firebase/src/helpers/export-lock.ts | 61 ++++++++++++ 4 files changed, 148 insertions(+), 37 deletions(-) create mode 100644 packages/firebase/src/helpers/export-lock.ts diff --git a/packages/apps/bling-erp/src/integration/export-order-to-bling.ts b/packages/apps/bling-erp/src/integration/export-order-to-bling.ts index 3b76b7488..b8043dc31 100644 --- a/packages/apps/bling-erp/src/integration/export-order-to-bling.ts +++ b/packages/apps/bling-erp/src/integration/export-order-to-bling.ts @@ -3,6 +3,7 @@ import { URLSearchParams } from 'url'; import ecomUtils from '@ecomplus/utils'; import api from '@cloudcommerce/api'; import { logger } from '@cloudcommerce/firebase/lib/config'; +import { tryLockExport, unlockExport } from '@cloudcommerce/firebase/lib/helpers/export-lock'; import { createBlingClient } from '../bling-auth/client'; import parseOrder from './parsers/order-to-bling'; import parseStatusToBling from './parsers/status-to-bling'; @@ -11,6 +12,9 @@ import getProductsBling from './helpers/get-products-bling'; import { getPaymentBling } from './helpers/payment-method'; import getStatusBling from './helpers/get-status-bling'; +const EXPORT_LOCKS_COLLECTION = 'blingExportedOrders'; +const CONCURRENT_EXPORT_WAIT_MS = 5000; + /* `canCreateNew` is a tri-state: - `true`: always create the order on Bling when not found; @@ -127,6 +131,7 @@ const exportOrderToBling = async ( } else if (data) { blingSavedOrder = data; } + let hasExportLock = false; if (blingSavedOrder) { blingOrderId = blingSavedOrder.id; @@ -148,6 +153,22 @@ const exportOrderToBling = async ( : String(order.number); } + /* Creating is not idempotent: the same order gets one event at checkout and + another on payment, and two executions can run them at the same time. Both + searched Bling before either wrote, so without this lock both create. */ + if (!blingOrderId) { + hasExportLock = await tryLockExport(EXPORT_LOCKS_COLLECTION, orderId); + if (!hasExportLock) { + logger.info(`${orderId} waiting concurrent export to Bling`); + await new Promise((resolve) => { + setTimeout(resolve, CONCURRENT_EXPORT_WAIT_MS); + }); + /* Starts over to find the order the winning execution created and go + on to the status update; `canCreateNew: false` can never create. */ + return exportOrderToBling(apiDoc, queueEntry, appData, false); + } + } + const customerIdBling = await getCustomerBling(bling, appData, order); if (!customerIdBling) { throw new Error('Bling Customer not found'); @@ -204,6 +225,10 @@ const exportOrderToBling = async ( await api.patch(`orders/${order._id}`, { metafields } as any).catch(logger.error); } } catch (err: any) { + if (hasExportLock) { + /* Nothing was created, so the lock must not block the next attempt. */ + await unlockExport(EXPORT_LOCKS_COLLECTION, orderId); + } if (err.response) { logger.warn(`Failed exporting order ${order._id}`, { blingOrder, diff --git a/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts b/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts index 8662df869..c9bf2bb5e 100644 --- a/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts +++ b/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts @@ -1,11 +1,15 @@ import type { Orders } from '@cloudcommerce/types'; import { logger } from '@cloudcommerce/firebase/lib/config'; +import { tryLockExport, unlockExport } from '@cloudcommerce/firebase/lib/helpers/export-lock'; import api from '@cloudcommerce/api'; import postTiny from './post-tiny-erp'; import parseStatusToTiny from './parsers/status-to-tiny'; import parseTinyStatus from './parsers/status-from-tiny'; import parseOrder from './parsers/order-to-tiny'; +const EXPORT_LOCKS_COLLECTION = 'tinyExportedOrders'; +const CONCURRENT_EXPORT_WAIT_MS = 5000; + export default async (apiDoc, queueEntry, appData, canCreateNew) => { const orderId = queueEntry.nextId; let order: Orders; @@ -52,39 +56,38 @@ export default async (apiDoc, queueEntry, appData, canCreateNew) => { logger.info(`${orderId} skipped with last fulfillment status from Tiny`); return null; } - logger.info(`${orderId} searching order ${order.number}`); - - let tinyData: { pedidos?: any }; - try { - tinyData = await postTiny('/pedidos.pesquisa.php', { - numeroEcommerce: String(order.number), - }); - } catch (err: any) { - const status = err.response && err.response.status; - if (status === 404) { - tinyData = {}; - } else { - logger.info(`${orderId} search on tiny ends with status ${status}`); - if (!status) { - logger.error(err); + const searchTinyOrder = async () => { + logger.info(`${orderId} searching order ${order.number}`); + let tinyData: { pedidos?: any }; + try { + tinyData = await postTiny('/pedidos.pesquisa.php', { + numeroEcommerce: String(order.number), + }); + } catch (err: any) { + const status = err.response && err.response.status; + if (status === 404) { + tinyData = {}; + } else { + logger.info(`${orderId} search on tiny ends with status ${status}`); + if (!status) { + logger.error(err); + } + throw err; } - throw err; } - } - const { pedidos } = tinyData; - let originalTinyOrder: undefined | Record; - if (Array.isArray(pedidos)) { - originalTinyOrder = pedidos + const { pedidos } = tinyData; + if (!Array.isArray(pedidos)) return undefined; + const foundEntry = pedidos .sort((a: { pedido: Record }, b: { pedido: Record }) => { return a.pedido.id < b.pedido.id ? 1 : -1; }) .find(({ pedido }) => { return order.number === Number(pedido.numero_ecommerce); }); - if (originalTinyOrder) { - originalTinyOrder = originalTinyOrder.pedido; - } - } + return foundEntry?.pedido as undefined | Record; + }; + + let originalTinyOrder = await searchTinyOrder(); if (!originalTinyOrder) { if (!canCreateNew) { @@ -113,24 +116,45 @@ export default async (apiDoc, queueEntry, appData, canCreateNew) => { break; } } - const tinyOrder = await parseOrder(order, appData); - try { - logger.info(`Posting ${orderId}`, { tinyOrder }); - } catch { - logger.info(`Posting ${orderId} (stringify failed)`); - } - return postTiny('/pedido.incluir.php', { - pedido: { - pedido: tinyOrder, - }, - }).then((response) => { + if (!(await tryLockExport(EXPORT_LOCKS_COLLECTION, orderId))) { + /* Another execution is creating this same order on Tiny right now: the + search above cannot see it yet, so creating it here would duplicate it. */ + logger.info(`${orderId} waiting concurrent export to Tiny`); + await new Promise((resolve) => { + setTimeout(resolve, CONCURRENT_EXPORT_WAIT_MS); + }); + originalTinyOrder = await searchTinyOrder(); + if (!originalTinyOrder) { + logger.warn(`${orderId} skipped while another execution exports it`); + return null; + } + /* Falls through to the status update with the order the winner created. */ + } else { + const tinyOrder = await parseOrder(order, appData); + try { + logger.info(`Posting ${orderId}`, { tinyOrder }); + } catch { + logger.info(`Posting ${orderId} (stringify failed)`); + } + let response: Record; + try { + response = await postTiny('/pedido.incluir.php', { + pedido: { + pedido: tinyOrder, + }, + }); + } catch (err) { + /* Nothing was created, so the lock must not block the next attempt. */ + await unlockExport(EXPORT_LOCKS_COLLECTION, orderId); + throw err; + } logger.info(`Created ${orderId} on Tiny`, { response }); const tinyErpOnNewOrder = global.$tinyErpOnNewOrder; if (tinyErpOnNewOrder && typeof tinyErpOnNewOrder === 'function') { return tinyErpOnNewOrder({ response, order, postTiny }); } return response; - }); + } } if (originalTinyOrder) { diff --git a/packages/firebase/package.json b/packages/firebase/package.json index 961a1c6cc..12a0340be 100644 --- a/packages/firebase/package.json +++ b/packages/firebase/package.json @@ -13,6 +13,7 @@ "./lib/env": "./lib/env.js", "./lib/helpers/pubsub": "./lib/helpers/pubsub.js", "./lib/helpers/firestore": "./lib/helpers/firestore.js", + "./lib/helpers/export-lock": "./lib/helpers/export-lock.js", "./lib/helpers/update-app-data": "./lib/helpers/update-app-data.js", "./lib/helpers/get-app-data": "./lib/helpers/get-app-data.js" }, diff --git a/packages/firebase/src/helpers/export-lock.ts b/packages/firebase/src/helpers/export-lock.ts new file mode 100644 index 000000000..f7870a780 --- /dev/null +++ b/packages/firebase/src/helpers/export-lock.ts @@ -0,0 +1,61 @@ +import { getFirestore, Timestamp } from 'firebase-admin/firestore'; +import { logger } from '../config'; + +/* +Locks the creation of an external record (such as an ERP order) to one execution. + +Resource events are published per modified fields, so the same order usually gets +two events within seconds (checkout, then payment). Each event runs on its own +function execution, and nothing stops two of them from running at the same time: +`maxInstances` is best-effort and is exceeded on concurrent cold starts. Checking +whether the record already exists on the ERP is not enough to dedupe, because +creating it takes seconds and both executions read "not found" before either writes. +*/ + +const DEFAULT_STALE_LOCK_MS = 1000 * 60 * 5; + +const ALREADY_EXISTS = 6; + +export const tryLockExport = async ( + collectionName: string, + resourceId: string, + staleLockMs = DEFAULT_STALE_LOCK_MS, +): Promise => { + const docRef = getFirestore().collection(collectionName).doc(resourceId); + try { + await docRef.create({ lockedAt: Timestamp.now() }); + return true; + } catch (err: any) { + if (err?.code !== ALREADY_EXISTS) throw err; + } + /* A lock is also left behind when the winning execution dies before releasing + it (timeout, instance eviction). Taking it over after `staleLockMs` keeps the + resource exportable, at the cost of a duplicate if the first attempt is somehow + still running - hence a window much longer than any export takes. */ + try { + const takenOver = await getFirestore().runTransaction(async (transaction) => { + const snapshot = await transaction.get(docRef); + const lockedAt = snapshot.get('lockedAt') as Timestamp | undefined; + if (lockedAt && Date.now() - lockedAt.toMillis() < staleLockMs) { + return false; + } + transaction.set(docRef, { lockedAt: Timestamp.now() }); + return true; + }); + if (takenOver) { + logger.warn(`Took over stale export lock ${collectionName}/${resourceId}`); + } + return takenOver; + } catch (err: any) { + logger.warn(`Failed checking export lock ${collectionName}/${resourceId}`, { err }); + return false; + } +}; + +export const unlockExport = async (collectionName: string, resourceId: string) => { + try { + await getFirestore().collection(collectionName).doc(resourceId).delete(); + } catch (err: any) { + logger.warn(`Failed releasing export lock ${collectionName}/${resourceId}`, { err }); + } +}; From 02e3c76cfa950ca8b2ac805107e7bd0f1dd60822 Mon Sep 17 00:00:00 2001 From: Vitor Date: Mon, 9 Nov 2026 18:22:48 -0300 Subject: [PATCH 2/2] =?UTF-8?q?fix(apps):=20Pedido=20n=C3=A3o=20se=20perde?= =?UTF-8?q?=20no=20Tiny=20e=20no=20Bling=20quando=20duas=20exporta=C3=A7?= =?UTF-8?q?=C3=B5es=20concorrem?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Responde às revisões da PR. A execução que perdia o lock esperava 5 s, procurava o pedido uma vez no ERP e saía como sucesso: o pedido ficava sem exportar quando o vencedor falhava ou demorava, e no Bling com número aleatório a atualização de status se perdia sempre. - `acquireExportLock` assume o ciclo do lock. Quem perde acompanha o doc no Firestore até o vencedor gravar o id criado (segue para a atualização de status com ele), liberar o lock (assume e cria) ou o lock ficar velho. Passado `maxWaitMs`, lança `ConcurrentExportError`, que o `afterQueue` rejeita para o evento ser reentregue, também na exportação automática. - O lock é liberado em qualquer falha antes do POST e em erro respondido pelo ERP. Sem resposta (timeout) fica até expirar, porque o pedido pode ter sido criado. - Erro do Firestore não bloqueia a exportação: ela segue sem lock, como antes. - Janela de lock velho por app, maior que o timeout da função (Tiny 2 min, Bling 6 min). O `eventMaxAgeMs` do Tiny sobe para 10 min para a reentrega não ser descartada. - O cron do Tiny não espera lock alheio. Coleções renomeadas para `tinyErpExportLocks` e `blingExportLocks`, com `expiresAt` para TTL. - Testes do helper com Firestore em memória. Co-Authored-By: Claude Opus 5.5 --- .../src/integration/after-bling-queue.ts | 12 + .../src/integration/export-order-to-bling.ts | 182 ++++++++------- .../src/integration/after-tiny-queue.ts | 12 + .../src/integration/export-order-to-tiny.ts | 59 +++-- .../apps/tiny-erp/src/tiny-erp-send-orders.ts | 23 +- packages/apps/tiny-erp/src/tiny-erp.ts | 5 +- packages/firebase/src/helpers/export-lock.ts | 209 +++++++++++++---- packages/firebase/tests/export-lock.test.ts | 212 ++++++++++++++++++ 8 files changed, 567 insertions(+), 147 deletions(-) create mode 100644 packages/firebase/tests/export-lock.test.ts diff --git a/packages/apps/bling-erp/src/integration/after-bling-queue.ts b/packages/apps/bling-erp/src/integration/after-bling-queue.ts index b839b485d..f00463f30 100644 --- a/packages/apps/bling-erp/src/integration/after-bling-queue.ts +++ b/packages/apps/bling-erp/src/integration/after-bling-queue.ts @@ -10,6 +10,18 @@ export default async ( payload: any, ) => { const isError = payload instanceof Error; + if (isError && (payload as any).isConcurrentExport) { + /* Outra execução segura o lock de exportação deste pedido por tempo demais. + Rejeitar faz o evento ser reentregue (`failurePolicy`) também na exportação + automática (fora da fila), e a nova tentativa acha o pedido exportado ou assume + o lock, em vez de terminar como falso sucesso sem nada no Bling. */ + logger.warn(payload.message); + return new Promise((resolve, reject) => { + setTimeout(() => { + reject(payload); + }, 2000); + }); + } const isImportation = !!queueEntry.action?.endsWith('importation'); const isQueued = !queueEntry.isNotQueued; const logs = appData.logs || []; diff --git a/packages/apps/bling-erp/src/integration/export-order-to-bling.ts b/packages/apps/bling-erp/src/integration/export-order-to-bling.ts index b8043dc31..aeffcfd6c 100644 --- a/packages/apps/bling-erp/src/integration/export-order-to-bling.ts +++ b/packages/apps/bling-erp/src/integration/export-order-to-bling.ts @@ -3,7 +3,10 @@ import { URLSearchParams } from 'url'; import ecomUtils from '@ecomplus/utils'; import api from '@cloudcommerce/api'; import { logger } from '@cloudcommerce/firebase/lib/config'; -import { tryLockExport, unlockExport } from '@cloudcommerce/firebase/lib/helpers/export-lock'; +import { + type ExportLock, + acquireExportLock, +} from '@cloudcommerce/firebase/lib/helpers/export-lock'; import { createBlingClient } from '../bling-auth/client'; import parseOrder from './parsers/order-to-bling'; import parseStatusToBling from './parsers/status-to-bling'; @@ -12,8 +15,13 @@ import getProductsBling from './helpers/get-products-bling'; import { getPaymentBling } from './helpers/payment-method'; import getStatusBling from './helpers/get-status-bling'; -const EXPORT_LOCKS_COLLECTION = 'blingExportedOrders'; -const CONCURRENT_EXPORT_WAIT_MS = 5000; +const EXPORT_LOCKS_COLLECTION = 'blingExportLocks'; +/* Maior que o timeout de 300s do `onStoreEvent`: uma execução lenta, mas viva, não +pode ter o lock tomado no meio da criação. */ +const EXPORT_LOCK_STALE_MS = 1000 * 60 * 6; +/* Depois de pegar o lock o vencedor ainda busca contato, forma de pagamento e +produtos antes do POST, com o client limitado a ~1 req/s. */ +const CONCURRENT_EXPORT_WAIT_MS = 1000 * 60; /* `canCreateNew` is a tri-state: @@ -131,7 +139,6 @@ const exportOrderToBling = async ( } else if (data) { blingSavedOrder = data; } - let hasExportLock = false; if (blingSavedOrder) { blingOrderId = blingSavedOrder.id; @@ -153,89 +160,108 @@ const exportOrderToBling = async ( : String(order.number); } - /* Creating is not idempotent: the same order gets one event at checkout and - another on payment, and two executions can run them at the same time. Both - searched Bling before either wrote, so without this lock both create. */ + /* Criar não é idempotente: o mesmo pedido recebe um evento no checkout e outro + no pagamento, e as duas execuções podem rodar ao mesmo tempo. As duas buscaram + no Bling antes de qualquer uma gravar, então sem o lock as duas criam. */ + let exportLock: ExportLock | undefined; + let isExportedConcurrently = false; if (!blingOrderId) { - hasExportLock = await tryLockExport(EXPORT_LOCKS_COLLECTION, orderId); - if (!hasExportLock) { - logger.info(`${orderId} waiting concurrent export to Bling`); - await new Promise((resolve) => { - setTimeout(resolve, CONCURRENT_EXPORT_WAIT_MS); - }); - /* Starts over to find the order the winning execution created and go - on to the status update; `canCreateNew: false` can never create. */ - return exportOrderToBling(apiDoc, queueEntry, appData, false); + const lockResult = await acquireExportLock(EXPORT_LOCKS_COLLECTION, orderId, { + staleLockMs: EXPORT_LOCK_STALE_MS, + maxWaitMs: CONCURRENT_EXPORT_WAIT_MS, + }); + if (lockResult.isLocked) { + exportLock = lockResult; + } else { + /* Criado por outra execução enquanto esta esperava. O id vem do lock porque + a busca acima não acha pedido com número aleatório nem com o `apiDoc` + lido antes de o vencedor gravar os metafields. */ + isExportedConcurrently = true; + blingOrderId = lockResult.exported.id; + logger.info(`${orderId} exported by a concurrent execution`, { blingOrderId }); } } - const customerIdBling = await getCustomerBling(bling, appData, order); - if (!customerIdBling) { - throw new Error('Bling Customer not found'); - } - /* Buscado só quando o pedido vai ser criado/atualizado de fato: mudança de - status de pedido já exportado não usa forma de pagamento nem itens. */ - const paymentTypeId = transaction - ? await getPaymentBling(bling, transaction, appData.parse_payment, order.payment_method_label) - : null; - const itemsBling = await getProductsBling(bling, order); - const blingOrder = parseOrder( - order, - blingOrderNumber, - blingStore, - appData, - customerIdBling, - paymentTypeId, - itemsBling, - blingSavedOrder, - ); - const endpoint = `/pedidos/vendas${blingOrderId ? `/${blingOrderId}` : ''}`; - const method = blingOrderId ? 'put' : 'post'; - logger.info(`[${method}]: ${endpoint} for ${order._id}`, { blingOrder, blingStatuses }); - try { - const { data: { data: savedData } } = await bling[method](endpoint, blingOrder); - logger.info(`Bling order ${method === 'put' ? 'upd' : 'cre'}ated successfully`); - if (savedData?.id) { - blingSavedOrder = blingOrder; - blingOrderId = savedData.id; - if (metafieldId) { - metafieldId.value = String(blingOrderId); - } else { - metafields.push({ - _id: ecomUtils.randomObjectId(), - namespace: 'bling', - field: 'bling:id', - value: String(blingOrderId), - }); + if (!isExportedConcurrently) { + let blingOrder: Record | undefined; + try { + const customerIdBling = await getCustomerBling(bling, appData, order); + if (!customerIdBling) { + throw new Error('Bling Customer not found'); } - /* Guardado para a busca por número funcionar quando o número do Bling - não é o da loja (`random_order_number` ou colisão de numeração). */ - if ( - blingOrderNumber - && !appData.disable_order_number - && !metafields.some(({ field }) => field === 'bling:numero') - ) { - metafields.push({ - _id: ecomUtils.randomObjectId(), - namespace: 'bling', - field: 'bling:numero', - value: String(blingOrderNumber), + /* Buscado só quando o pedido vai ser criado/atualizado de fato: mudança de + status de pedido já exportado não usa forma de pagamento nem itens. */ + const paymentTypeId = transaction + ? await getPaymentBling( + bling, + transaction, + appData.parse_payment, + order.payment_method_label, + ) + : null; + const itemsBling = await getProductsBling(bling, order); + blingOrder = parseOrder( + order, + blingOrderNumber, + blingStore, + appData, + customerIdBling, + paymentTypeId, + itemsBling, + blingSavedOrder, + ); + const endpoint = `/pedidos/vendas${blingOrderId ? `/${blingOrderId}` : ''}`; + const method = blingOrderId ? 'put' : 'post'; + logger.info(`[${method}]: ${endpoint} for ${order._id}`, { blingOrder, blingStatuses }); + exportLock?.setSending(); + const { data: { data: savedData } } = await bling[method](endpoint, blingOrder); + logger.info(`Bling order ${method === 'put' ? 'upd' : 'cre'}ated successfully`); + if (savedData?.id) { + blingSavedOrder = blingOrder; + blingOrderId = savedData.id; + if (metafieldId) { + metafieldId.value = String(blingOrderId); + } else { + metafields.push({ + _id: ecomUtils.randomObjectId(), + namespace: 'bling', + field: 'bling:id', + value: String(blingOrderId), + }); + } + /* Guardado para a busca por número funcionar quando o número do Bling + não é o da loja (`random_order_number` ou colisão de numeração). */ + if ( + blingOrderNumber + && !appData.disable_order_number + && !metafields.some(({ field }) => field === 'bling:numero') + ) { + metafields.push({ + _id: ecomUtils.randomObjectId(), + namespace: 'bling', + field: 'bling:numero', + value: String(blingOrderNumber), + }); + } + await api.patch(`orders/${order._id}`, { metafields } as any).catch(logger.error); + } + } catch (err: any) { + /* O lock é liberado para a próxima tentativa quando o pedido certamente não + foi criado: falha antes do POST (contato, forma de pagamento, produtos) ou + erro respondido pelo Bling. Sem resposta (timeout) o pedido pode ter sido + criado, e o lock fica até expirar. */ + await exportLock?.fail(err); + if (err.response) { + logger.warn(`Failed exporting order ${order._id}`, { + blingOrder, + response: err.response.data, }); } - await api.patch(`orders/${order._id}`, { metafields } as any).catch(logger.error); - } - } catch (err: any) { - if (hasExportLock) { - /* Nothing was created, so the lock must not block the next attempt. */ - await unlockExport(EXPORT_LOCKS_COLLECTION, orderId); + throw err; } - if (err.response) { - logger.warn(`Failed exporting order ${order._id}`, { - blingOrder, - response: err.response.data, - }); + if (exportLock && blingOrderId) { + await exportLock.complete({ id: blingOrderId, numero: blingOrderNumber }); } - throw err; } } diff --git a/packages/apps/tiny-erp/src/integration/after-tiny-queue.ts b/packages/apps/tiny-erp/src/integration/after-tiny-queue.ts index 4f566f4d2..8bc40024f 100644 --- a/packages/apps/tiny-erp/src/integration/after-tiny-queue.ts +++ b/packages/apps/tiny-erp/src/integration/after-tiny-queue.ts @@ -9,6 +9,18 @@ export default async ( payload: any, ) => { const isError = payload instanceof Error; + if (isError && (payload as any).isConcurrentExport) { + /* Another execution holds the export lock of this order for too long. Rejecting + gets the event redelivered (`failurePolicy`) even for automatic exports (not + queued), and the retry finds the order exported or takes over the lock, instead + of ending as a false success with nothing sent to Tiny. */ + logger.warn(payload.message); + return new Promise((resolve, reject) => { + setTimeout(() => { + reject(payload); + }, 2000); + }); + } const isImportation = !!queueEntry.action?.endsWith('importation'); const isQueued = !queueEntry.isNotQueued; const logs = appData.logs || []; diff --git a/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts b/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts index c9bf2bb5e..d2007d21e 100644 --- a/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts +++ b/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts @@ -1,14 +1,24 @@ import type { Orders } from '@cloudcommerce/types'; import { logger } from '@cloudcommerce/firebase/lib/config'; -import { tryLockExport, unlockExport } from '@cloudcommerce/firebase/lib/helpers/export-lock'; +import { acquireExportLock } from '@cloudcommerce/firebase/lib/helpers/export-lock'; import api from '@cloudcommerce/api'; import postTiny from './post-tiny-erp'; import parseStatusToTiny from './parsers/status-to-tiny'; import parseTinyStatus from './parsers/status-from-tiny'; import parseOrder from './parsers/order-to-tiny'; -const EXPORT_LOCKS_COLLECTION = 'tinyExportedOrders'; -const CONCURRENT_EXPORT_WAIT_MS = 5000; +const EXPORT_LOCKS_COLLECTION = 'tinyErpExportLocks'; +/* Longer than anything holding the lock can live: the events function times out +in 60s, and the cron holds it for a single order (the POST times out in 30s). */ +const EXPORT_LOCK_STALE_MS = 1000 * 60 * 2; +const CONCURRENT_EXPORT_WAIT_MS = 20000; + +/* `pedido.incluir.php` returns the created order in `registros.registro` */ +const getCreatedTinyId = (response: Record) => { + const { registros } = response; + const registro = (Array.isArray(registros) ? registros[0] : registros)?.registro; + return registro?.id as string | number | undefined; +}; export default async (apiDoc, queueEntry, appData, canCreateNew) => { const orderId = queueEntry.nextId; @@ -116,39 +126,46 @@ export default async (apiDoc, queueEntry, appData, canCreateNew) => { break; } } - if (!(await tryLockExport(EXPORT_LOCKS_COLLECTION, orderId))) { - /* Another execution is creating this same order on Tiny right now: the - search above cannot see it yet, so creating it here would duplicate it. */ - logger.info(`${orderId} waiting concurrent export to Tiny`); - await new Promise((resolve) => { - setTimeout(resolve, CONCURRENT_EXPORT_WAIT_MS); - }); - originalTinyOrder = await searchTinyOrder(); + const exportLock = await acquireExportLock(EXPORT_LOCKS_COLLECTION, orderId, { + staleLockMs: EXPORT_LOCK_STALE_MS, + maxWaitMs: queueEntry.concurrentExportWaitMs ?? CONCURRENT_EXPORT_WAIT_MS, + }); + if (!exportLock.isLocked) { + /* Created by a concurrent execution while this one waited, the search above + could not see it: falls through to the status update on that order. */ + const { exported } = exportLock; + originalTinyOrder = exported.id + ? { id: exported.id, situacao: exported.situacao } + : await searchTinyOrder(); if (!originalTinyOrder) { - logger.warn(`${orderId} skipped while another execution exports it`); + logger.warn(`${orderId} exported by a concurrent execution but not found`); return null; } - /* Falls through to the status update with the order the winner created. */ } else { - const tinyOrder = await parseOrder(order, appData); - try { - logger.info(`Posting ${orderId}`, { tinyOrder }); - } catch { - logger.info(`Posting ${orderId} (stringify failed)`); - } + let tinyOrder: Record; let response: Record; try { + tinyOrder = await parseOrder(order, appData); + try { + logger.info(`Posting ${orderId}`, { tinyOrder }); + } catch { + logger.info(`Posting ${orderId} (stringify failed)`); + } + exportLock.setSending(); response = await postTiny('/pedido.incluir.php', { pedido: { pedido: tinyOrder, }, }); } catch (err) { - /* Nothing was created, so the lock must not block the next attempt. */ - await unlockExport(EXPORT_LOCKS_COLLECTION, orderId); + await exportLock.fail(err); throw err; } logger.info(`Created ${orderId} on Tiny`, { response }); + await exportLock.complete({ + id: getCreatedTinyId(response), + situacao: tinyOrder.situacao, + }); const tinyErpOnNewOrder = global.$tinyErpOnNewOrder; if (tinyErpOnNewOrder && typeof tinyErpOnNewOrder === 'function') { return tinyErpOnNewOrder({ response, order, postTiny }); diff --git a/packages/apps/tiny-erp/src/tiny-erp-send-orders.ts b/packages/apps/tiny-erp/src/tiny-erp-send-orders.ts index 8156d915b..0559b5946 100644 --- a/packages/apps/tiny-erp/src/tiny-erp-send-orders.ts +++ b/packages/apps/tiny-erp/src/tiny-erp-send-orders.ts @@ -63,7 +63,9 @@ export const sendWaitingOrders = async () => { const order = orders[i] as Orders; if (!order.number) continue; try { - const queueEntry = { nextId: order._id }; + /* An order locked by a concurrent export is left to it (and to the next + run) instead of holding the sequential loop while it finishes. */ + const queueEntry = { nextId: order._id, concurrentExportWaitMs: 0 }; // eslint-disable-next-line no-await-in-loop const result = await exportOrderToTiny(order, queueEntry, appData, true); if (result && typeof result === 'object' && 'registros' in result) { @@ -72,13 +74,18 @@ export const sendWaitingOrders = async () => { skippedCount += 1; } } catch (error: any) { - const err = new Error(`Failed sending order ${order.number} to Tiny`); - logger.error(err, { - orderId: order._id, - orderNumber: order.number, - request: error.config, - response: error.response?.data, - }); + if (error.isConcurrentExport) { + logger.info(`Order ${order.number} skipped while exported concurrently`); + skippedCount += 1; + } else { + const err = new Error(`Failed sending order ${order.number} to Tiny`); + logger.error(err, { + orderId: order._id, + orderNumber: order.number, + request: error.config, + response: error.response?.data, + }); + } } // Rate limit: 150ms between requests // eslint-disable-next-line no-await-in-loop diff --git a/packages/apps/tiny-erp/src/tiny-erp.ts b/packages/apps/tiny-erp/src/tiny-erp.ts index f99c6e0f2..11d725fbe 100644 --- a/packages/apps/tiny-erp/src/tiny-erp.ts +++ b/packages/apps/tiny-erp/src/tiny-erp.ts @@ -15,7 +15,10 @@ export const tinyerp = { onStoreEvent: createAppEventsFunction( 'tinyErp', handleApiEvent, - { memory: '512MB' }, + /* With the default max age (60s), an order event redelivered after waiting on a + concurrent export (`ConcurrentExportError`) would be dropped before an abandoned + export lock gets stale (2 min) and can be taken over. */ + { memory: '512MB', eventMaxAgeMs: 1000 * 60 * 10 }, ), webhook: functions diff --git a/packages/firebase/src/helpers/export-lock.ts b/packages/firebase/src/helpers/export-lock.ts index f7870a780..ea029ba80 100644 --- a/packages/firebase/src/helpers/export-lock.ts +++ b/packages/firebase/src/helpers/export-lock.ts @@ -1,3 +1,4 @@ +import { randomUUID } from 'node:crypto'; import { getFirestore, Timestamp } from 'firebase-admin/firestore'; import { logger } from '../config'; @@ -10,52 +11,182 @@ function execution, and nothing stops two of them from running at the same time: `maxInstances` is best-effort and is exceeded on concurrent cold starts. Checking whether the record already exists on the ERP is not enough to dedupe, because creating it takes seconds and both executions read "not found" before either writes. + +The lock doc is the source of truth for the executions that lose it: they wait on +it, not on the ERP, until the winner either saves what it created (`complete`) or +releases the lock because nothing was created (`fail`), in which case the next one +takes the lock and creates the record itself. + +Docs carry `expiresAt` for a Firestore TTL policy, they are useless after a day. */ -const DEFAULT_STALE_LOCK_MS = 1000 * 60 * 5; +const DEFAULT_POLL_INTERVAL_MS = 2000; +const LOCK_TTL_MS = 1000 * 60 * 60 * 24; + +export type ExportLockOptions = { + /* + A lock without a finished export older than this is taken over, so a record is + not left unexported forever when the winning execution dies between taking the + lock and creating it (timeout, instance eviction). It must be longer than an + execution can hold the lock (the timeout of the function taking it), otherwise + a slow but live execution is taken over and the record is duplicated. + A finished export older than this is taken over as well, it is how a record + deleted on the ERP on purpose can be exported again. + */ + staleLockMs: number, + // How long to wait on a concurrent execution before giving up + maxWaitMs: number, + pollIntervalMs?: number, +}; + +export type ExportLock = { + isLocked: true, + // To call right before sending the request that creates the record + setSending: () => void, + // Saves what concurrent executions need to go on without creating (the ERP id) + complete: (exported: Record) => Promise, + /* + Releases the lock so the next execution can create the record, unless the create + request was sent and failed without a response (timeout, network): the record may + have been created anyway, so the lock is kept until stale to avoid a duplicate. + */ + fail: (err: any) => Promise, +}; + +export type ConcurrentExport = { + isLocked: false, + // Saved by the concurrent execution that created the record + exported: Record, +}; + +/* +Thrown when a concurrent execution holds the lock for longer than `maxWaitMs`. +It is meant to be retried later (event redelivery), never handled as success. +*/ +export class ConcurrentExportError extends Error { + isConcurrentExport = true; + + constructor(lockPath: string) { + super(`Exportação de ${lockPath} em andamento em outra execução`); + this.name = 'ConcurrentExportError'; + } +} -const ALREADY_EXISTS = 6; +const sleep = (ms: number) => new Promise((resolve) => { + setTimeout(resolve, ms); +}); -export const tryLockExport = async ( +export const acquireExportLock = async ( collectionName: string, resourceId: string, - staleLockMs = DEFAULT_STALE_LOCK_MS, -): Promise => { - const docRef = getFirestore().collection(collectionName).doc(resourceId); - try { - await docRef.create({ lockedAt: Timestamp.now() }); - return true; - } catch (err: any) { - if (err?.code !== ALREADY_EXISTS) throw err; - } - /* A lock is also left behind when the winning execution dies before releasing - it (timeout, instance eviction). Taking it over after `staleLockMs` keeps the - resource exportable, at the cost of a duplicate if the first attempt is somehow - still running - hence a window much longer than any export takes. */ - try { - const takenOver = await getFirestore().runTransaction(async (transaction) => { - const snapshot = await transaction.get(docRef); - const lockedAt = snapshot.get('lockedAt') as Timestamp | undefined; - if (lockedAt && Date.now() - lockedAt.toMillis() < staleLockMs) { - return false; + { + staleLockMs, + maxWaitMs, + pollIntervalMs = DEFAULT_POLL_INTERVAL_MS, + }: ExportLockOptions, +): Promise => { + const firestore = getFirestore(); + const docRef = firestore.collection(collectionName).doc(resourceId); + const lockPath = `${collectionName}/${resourceId}`; + const lockId = randomUUID(); + /* Without Firestore the export goes on unlocked, as before the lock existed: + a possible duplicate is better than an order that never reaches the ERP. */ + let isTracked = true; + let isSending = false; + + const lock: ExportLock = { + isLocked: true, + setSending() { + isSending = true; + }, + async complete(exported) { + if (!isTracked) return; + const exportedData: Record = {}; + Object.keys(exported).forEach((field) => { + if (exported[field] !== undefined) exportedData[field] = exported[field]; + }); + try { + await docRef.set({ + exported: exportedData, + exportedAt: Timestamp.now(), + }, { merge: true }); + } catch (err: any) { + logger.warn(`Failed saving export lock ${lockPath}`, { err }); } - transaction.set(docRef, { lockedAt: Timestamp.now() }); - return true; - }); - if (takenOver) { - logger.warn(`Took over stale export lock ${collectionName}/${resourceId}`); - } - return takenOver; - } catch (err: any) { - logger.warn(`Failed checking export lock ${collectionName}/${resourceId}`, { err }); - return false; - } -}; + }, + async fail(err) { + if (!isTracked) return; + if (isSending && !err?.response) { + logger.warn(`Keeping export lock ${lockPath}, record may have been created`, { + err, + }); + return; + } + try { + await firestore.runTransaction(async (transaction) => { + const snapshot = await transaction.get(docRef); + // The lock may have been taken over meanwhile, it is not ours to release + if (snapshot.get('lockId') === lockId) { + transaction.delete(docRef); + } + }); + } catch (_err: any) { + logger.warn(`Failed releasing export lock ${lockPath}`, { err: _err }); + } + }, + }; -export const unlockExport = async (collectionName: string, resourceId: string) => { - try { - await getFirestore().collection(collectionName).doc(resourceId).delete(); - } catch (err: any) { - logger.warn(`Failed releasing export lock ${collectionName}/${resourceId}`, { err }); + const startedAt = Date.now(); + let isWaitLogged = false; + for (;;) { + let state: 'locked' | 'takenOver' | 'busy' | { exported: Record }; + try { + // eslint-disable-next-line no-await-in-loop + state = await firestore.runTransaction(async (transaction) => { + const snapshot = await transaction.get(docRef); + const now = Date.now(); + const lockData = { + lockId, + lockedAt: Timestamp.fromMillis(now), + expiresAt: Timestamp.fromMillis(now + LOCK_TTL_MS), + }; + if (!snapshot.exists) { + transaction.create(docRef, lockData); + return 'locked'; + } + const exported = snapshot.get('exported') as Record | undefined; + const lockedAt = (snapshot.get('exportedAt') || snapshot.get('lockedAt')) as + Timestamp | undefined; + if (!lockedAt || now - lockedAt.toMillis() >= staleLockMs) { + transaction.set(docRef, lockData); + return 'takenOver'; + } + return exported ? { exported } : 'busy'; + }); + } catch (err: any) { + logger.error(err); + logger.warn(`Exporting ${lockPath} unlocked, export lock is unavailable`); + isTracked = false; + return lock; + } + if (state === 'locked') { + return lock; + } + if (state === 'takenOver') { + logger.warn(`Took over stale export lock ${lockPath}`); + return lock; + } + if (state !== 'busy') { + return { isLocked: false, exported: state.exported }; + } + if (Date.now() - startedAt + pollIntervalMs > maxWaitMs) { + throw new ConcurrentExportError(lockPath); + } + if (!isWaitLogged) { + logger.info(`Waiting concurrent export ${lockPath}`); + isWaitLogged = true; + } + // eslint-disable-next-line no-await-in-loop + await sleep(pollIntervalMs); } }; diff --git a/packages/firebase/tests/export-lock.test.ts b/packages/firebase/tests/export-lock.test.ts new file mode 100644 index 000000000..ce5f569d0 --- /dev/null +++ b/packages/firebase/tests/export-lock.test.ts @@ -0,0 +1,212 @@ +import { + describe, + test, + expect, + vi, + beforeEach, +} from 'vitest'; +import { Timestamp } from 'firebase-admin/firestore'; +import { + acquireExportLock, + ConcurrentExportError, + type ExportLock, +} from '../src/helpers/export-lock'; + +/* +In-memory Firestore with the subset used by the lock. Transactions run one at a +time, like the pessimistic locking of the server SDK, and their writes are only +applied when the callback resolves. +*/ +const { db } = vi.hoisted(() => { + const docs = new Map>(); + const state = { isDown: false }; + let queue: Promise = Promise.resolve(); + const checkUp = () => { + if (state.isDown) throw new Error('14 UNAVAILABLE'); + }; + const docRef = (path: string) => ({ + path, + async set(data: Record, opts?: { merge?: boolean }) { + checkUp(); + docs.set(path, opts?.merge ? { ...docs.get(path), ...data } : { ...data }); + }, + }); + const firestore = { + collection: (name: string) => ({ doc: (id: string) => docRef(`${name}/${id}`) }), + runTransaction(fn: (transaction: any) => Promise) { + const run = queue.then(async () => { + checkUp(); + const writes: Array<() => void> = []; + const result = await fn({ + async get({ path }: { path: string }) { + const data = docs.get(path); + return { exists: Boolean(data), get: (field: string) => data?.[field] }; + }, + create({ path }: { path: string }, data: Record) { + writes.push(() => { + if (docs.has(path)) throw new Error('6 ALREADY_EXISTS'); + docs.set(path, { ...data }); + }); + }, + set({ path }: { path: string }, data: Record) { + writes.push(() => docs.set(path, { ...data })); + }, + delete({ path }: { path: string }) { + writes.push(() => docs.delete(path)); + }, + }); + writes.forEach((write) => write()); + return result; + }); + queue = run.catch(() => null); + return run; + }, + }; + return { db: { docs, state, firestore } }; +}); + +vi.mock('firebase-admin/firestore', async (importOriginal) => ({ + ...(await importOriginal()), + getFirestore: () => db.firestore, +})); +vi.mock('../src/config', () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +const coll = 'testExportLocks'; +const docPath = `${coll}/order1`; +const opts = { staleLockMs: 60000, maxWaitMs: 400, pollIntervalMs: 10 }; +const acquire = (options: Partial = {}) => { + return acquireExportLock(coll, 'order1', { ...opts, ...options }); +}; +const minutesAgo = (minutes: number) => Timestamp.fromMillis(Date.now() - minutes * 60000); +const sleep = (ms: number) => new Promise((resolve) => { + setTimeout(resolve, ms); +}); + +const acquireLocked = async (options?: Partial) => { + const lock = await acquire(options); + if (!lock.isLocked) throw new Error('Expected to take the lock'); + return lock as ExportLock; +}; + +describe('Export lock', () => { + beforeEach(() => { + db.docs.clear(); + db.state.isDown = false; + }); + + test('takes a free lock, with expiresAt for the TTL policy', async () => { + const lock = await acquire(); + expect(lock.isLocked).toBe(true); + const doc = db.docs.get(docPath); + expect(doc?.lockedAt).toBeInstanceOf(Timestamp); + expect(doc?.expiresAt.toMillis()).toBeGreaterThan(Date.now() + 60 * 60000); + }); + + test('locks per resource', async () => { + await acquireLocked(); + const other = await acquireExportLock(coll, 'order2', opts); + expect(other.isLocked).toBe(true); + }); + + test('concurrent execution waits and gets what the winner exported', async () => { + const first = acquire(); + const second = acquire(); + const winner = await first; + if (!winner.isLocked) throw new Error('First execution should win'); + await sleep(30); + await winner.complete({ id: 123, numero: undefined }); + expect(await second).toEqual({ isLocked: false, exported: { id: 123 } }); + }); + + test('concurrent execution takes the lock when the winner fails before sending', async () => { + const winner = await acquireLocked(); + const second = acquire(); + await sleep(30); + await winner.fail(new Error('Bling Customer not found')); + expect((await second).isLocked).toBe(true); + }); + + test('released when the ERP answered the create request with an error', async () => { + const winner = await acquireLocked(); + winner.setSending(); + await winner.fail(Object.assign(new Error('Tiny error 6'), { response: { status: 503 } })); + expect(db.docs.has(docPath)).toBe(false); + expect((await acquire()).isLocked).toBe(true); + }); + + test('kept when the create request failed with no response', async () => { + const winner = await acquireLocked(); + winner.setSending(); + await winner.fail(Object.assign(new Error('timeout of 10000ms exceeded'), { + code: 'ECONNABORTED', + })); + expect(db.docs.has(docPath)).toBe(true); + await expect(acquire({ maxWaitMs: 50 })).rejects.toBeInstanceOf(ConcurrentExportError); + }); + + test('gives up with a retryable error after maxWaitMs', async () => { + await acquireLocked(); + const startedAt = Date.now(); + const err = await acquire({ maxWaitMs: 100 }).catch((_err) => _err); + expect(err).toBeInstanceOf(ConcurrentExportError); + expect(err.isConcurrentExport).toBe(true); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(80); + }); + + test('does not wait at all with maxWaitMs 0', async () => { + await acquireLocked(); + const startedAt = Date.now(); + await expect(acquire({ maxWaitMs: 0 })).rejects.toBeInstanceOf(ConcurrentExportError); + expect(Date.now() - startedAt).toBeLessThan(50); + }); + + test('takes over a stale lock, then the renewed lock is busy', async () => { + db.docs.set(docPath, { lockId: 'dead', lockedAt: minutesAgo(2) }); + expect((await acquire()).isLocked).toBe(true); + expect(db.docs.get(docPath)?.lockId).not.toBe('dead'); + await expect(acquire({ maxWaitMs: 0 })).rejects.toBeInstanceOf(ConcurrentExportError); + }); + + test('takes over a doc without lockedAt instead of locking forever', async () => { + db.docs.set(docPath, { foo: 'bar' }); + expect((await acquire()).isLocked).toBe(true); + }); + + test('recent export is returned, old export is taken over to export again', async () => { + db.docs.set(docPath, { + lockId: 'done', + lockedAt: minutesAgo(0.5), + exported: { id: 1 }, + exportedAt: Timestamp.now(), + }); + expect(await acquire()).toEqual({ isLocked: false, exported: { id: 1 } }); + db.docs.set(docPath, { + lockId: 'done', + lockedAt: minutesAgo(5), + exported: { id: 1 }, + exportedAt: minutesAgo(2), + }); + expect((await acquire()).isLocked).toBe(true); + }); + + test('former holder of a taken over lock does not release the new one', async () => { + const formerHolder = await acquireLocked(); + db.docs.get(docPath)!.lockedAt = minutesAgo(2); + await acquireLocked(); + await formerHolder.fail(new Error('Bling Customer not found')); + expect(db.docs.has(docPath)).toBe(true); + await expect(acquire({ maxWaitMs: 0 })).rejects.toBeInstanceOf(ConcurrentExportError); + }); + + test('exports unlocked when Firestore is unavailable', async () => { + db.state.isDown = true; + const lock = await acquireLocked(); + lock.setSending(); + await expect(lock.complete({ id: 1 })).resolves.toBeUndefined(); + await expect(lock.fail(new Error('timeout'))).resolves.toBeUndefined(); + db.state.isDown = false; + expect(db.docs.size).toBe(0); + }); +});