diff --git a/lib/auth.js b/lib/auth.js index b3895d6..1c502b4 100644 --- a/lib/auth.js +++ b/lib/auth.js @@ -2,8 +2,6 @@ const fs = require('fs'); const path = require('path'); const {google} = require('googleapis'); const {applyRateLimitBuffer} = require('./rate'); -const pThrottleModule = require('p-throttle'); -const pThrottle = pThrottleModule.default || pThrottleModule; function loadApiKeys(config) { const apiKeys = []; @@ -58,7 +56,9 @@ function loadServiceAccounts(config) { function createServiceAccountBatchFetcher(config) { const accounts = loadServiceAccounts(config); if (accounts.length === 0) return null; - + // Build raw fetchers and track next available timestamps to allow + // selecting a credential whose slot is available now. If none are + // immediately available, wait the minimum required time. const fetchers = accounts.map(({path: credentialsPath, rateLimitPerMinute}) => { const resolvedPath = path.isAbsolute(credentialsPath) ? credentialsPath : path.resolve(process.cwd(), credentialsPath); if (!fs.existsSync(resolvedPath)) { @@ -82,7 +82,8 @@ function createServiceAccountBatchFetcher(config) { const bufferedRateLimit = applyRateLimitBuffer(rateLimitPerMinute); const interval = Math.ceil(60000 / bufferedRateLimit); - const throttled = pThrottle({limit: 1, interval})(async (documentId, sheetNames) => { + + const fetchRaw = async (documentId, sheetNames) => { const sheets = google.sheets({version: 'v4', auth: authClient}); const request = {spreadsheetId: documentId, ranges: sheetNames}; const response = await sheets.spreadsheets.values.batchGet(request); @@ -93,16 +94,47 @@ function createServiceAccountBatchFetcher(config) { rowsBySheet[sheetName] = valueRanges[i]?.values || []; } return rowsBySheet; - }); + }; - return {authClient, fetch: throttled}; + return {authClient, fetchRaw, interval, nextAvailable: 0}; }); let currentIndex = 0; + + const sleep = ms => new Promise(resolve => setTimeout(resolve, ms)); + return async (documentId, sheetNames) => { - const worker = fetchers[currentIndex]; - currentIndex = (currentIndex + 1) % fetchers.length; - return worker.fetch(documentId, sheetNames); + const now = () => Date.now(); + + // Try to find an available worker without blocking; check each worker once + for (let i = 0; i < fetchers.length; i += 1) { + const idx = (currentIndex + i) % fetchers.length; + const worker = fetchers[idx]; + if (now() >= worker.nextAvailable) { + currentIndex = (idx + 1) % fetchers.length; + worker.nextAvailable = now() + worker.interval; + return worker.fetchRaw(documentId, sheetNames); + } + } + + // If none available immediately, pick the earliest one and wait until it's free + let earliestIdx = 0; + let earliestTime = fetchers[0].nextAvailable; + for (let i = 1; i < fetchers.length; i += 1) { + if (fetchers[i].nextAvailable < earliestTime) { + earliestTime = fetchers[i].nextAvailable; + earliestIdx = i; + } + } + + const waitMs = Math.max(0, earliestTime - now()); + if (waitMs > 0) await sleep(waitMs); + + // After waiting, reserve the slot and call + const worker = fetchers[earliestIdx]; + currentIndex = (earliestIdx + 1) % fetchers.length; + worker.nextAvailable = now() + worker.interval; + return worker.fetchRaw(documentId, sheetNames); }; } @@ -111,11 +143,11 @@ function createApiKeyBatchFetcher(config) { if (apiKeys.length === 0) { return null; } - const fetchers = apiKeys.map(({key, rateLimitPerMinute}) => { const bufferedRateLimit = applyRateLimitBuffer(rateLimitPerMinute); const interval = Math.ceil(60000 / bufferedRateLimit); - const throttled = pThrottle({limit: 1, interval})(async (documentId, sheetNames) => { + + const fetchRaw = async (documentId, sheetNames) => { const sheets = google.sheets({version: 'v4'}); const request = {spreadsheetId: documentId, ranges: sheetNames, key}; const response = await sheets.spreadsheets.values.batchGet(request); @@ -126,15 +158,44 @@ function createApiKeyBatchFetcher(config) { rowsBySheet[sheetName] = valueRanges[i]?.values || []; } return rowsBySheet; - }); - return {key, fetch: throttled}; + }; + + return {key, fetchRaw, interval, nextAvailable: 0}; }); let currentIndex = 0; + + const sleep = ms => new Promise(resolve => setTimeout(resolve, ms)); + return async (documentId, sheetNames) => { - const worker = fetchers[currentIndex]; - currentIndex = (currentIndex + 1) % fetchers.length; - return worker.fetch(documentId, sheetNames); + const now = () => Date.now(); + + for (let i = 0; i < fetchers.length; i += 1) { + const idx = (currentIndex + i) % fetchers.length; + const worker = fetchers[idx]; + if (now() >= worker.nextAvailable) { + currentIndex = (idx + 1) % fetchers.length; + worker.nextAvailable = now() + worker.interval; + return worker.fetchRaw(documentId, sheetNames); + } + } + + let earliestIdx = 0; + let earliestTime = fetchers[0].nextAvailable; + for (let i = 1; i < fetchers.length; i += 1) { + if (fetchers[i].nextAvailable < earliestTime) { + earliestTime = fetchers[i].nextAvailable; + earliestIdx = i; + } + } + + const waitMs = Math.max(0, earliestTime - now()); + if (waitMs > 0) await sleep(waitMs); + + const worker = fetchers[earliestIdx]; + currentIndex = (earliestIdx + 1) % fetchers.length; + worker.nextAvailable = now() + worker.interval; + return worker.fetchRaw(documentId, sheetNames); }; }