Update to enable round robin on keys and accounts so updates won't get blocked by low frequencies.
This commit is contained in:
+77
-16
@@ -2,8 +2,6 @@ const fs = require('fs');
|
|||||||
const path = require('path');
|
const path = require('path');
|
||||||
const {google} = require('googleapis');
|
const {google} = require('googleapis');
|
||||||
const {applyRateLimitBuffer} = require('./rate');
|
const {applyRateLimitBuffer} = require('./rate');
|
||||||
const pThrottleModule = require('p-throttle');
|
|
||||||
const pThrottle = pThrottleModule.default || pThrottleModule;
|
|
||||||
|
|
||||||
function loadApiKeys(config) {
|
function loadApiKeys(config) {
|
||||||
const apiKeys = [];
|
const apiKeys = [];
|
||||||
@@ -58,7 +56,9 @@ function loadServiceAccounts(config) {
|
|||||||
function createServiceAccountBatchFetcher(config) {
|
function createServiceAccountBatchFetcher(config) {
|
||||||
const accounts = loadServiceAccounts(config);
|
const accounts = loadServiceAccounts(config);
|
||||||
if (accounts.length === 0) return null;
|
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 fetchers = accounts.map(({path: credentialsPath, rateLimitPerMinute}) => {
|
||||||
const resolvedPath = path.isAbsolute(credentialsPath) ? credentialsPath : path.resolve(process.cwd(), credentialsPath);
|
const resolvedPath = path.isAbsolute(credentialsPath) ? credentialsPath : path.resolve(process.cwd(), credentialsPath);
|
||||||
if (!fs.existsSync(resolvedPath)) {
|
if (!fs.existsSync(resolvedPath)) {
|
||||||
@@ -82,7 +82,8 @@ function createServiceAccountBatchFetcher(config) {
|
|||||||
|
|
||||||
const bufferedRateLimit = applyRateLimitBuffer(rateLimitPerMinute);
|
const bufferedRateLimit = applyRateLimitBuffer(rateLimitPerMinute);
|
||||||
const interval = Math.ceil(60000 / bufferedRateLimit);
|
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 sheets = google.sheets({version: 'v4', auth: authClient});
|
||||||
const request = {spreadsheetId: documentId, ranges: sheetNames};
|
const request = {spreadsheetId: documentId, ranges: sheetNames};
|
||||||
const response = await sheets.spreadsheets.values.batchGet(request);
|
const response = await sheets.spreadsheets.values.batchGet(request);
|
||||||
@@ -93,16 +94,47 @@ function createServiceAccountBatchFetcher(config) {
|
|||||||
rowsBySheet[sheetName] = valueRanges[i]?.values || [];
|
rowsBySheet[sheetName] = valueRanges[i]?.values || [];
|
||||||
}
|
}
|
||||||
return rowsBySheet;
|
return rowsBySheet;
|
||||||
});
|
};
|
||||||
|
|
||||||
return {authClient, fetch: throttled};
|
return {authClient, fetchRaw, interval, nextAvailable: 0};
|
||||||
});
|
});
|
||||||
|
|
||||||
let currentIndex = 0;
|
let currentIndex = 0;
|
||||||
|
|
||||||
|
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
|
||||||
|
|
||||||
return async (documentId, sheetNames) => {
|
return async (documentId, sheetNames) => {
|
||||||
const worker = fetchers[currentIndex];
|
const now = () => Date.now();
|
||||||
currentIndex = (currentIndex + 1) % fetchers.length;
|
|
||||||
return worker.fetch(documentId, sheetNames);
|
// 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) {
|
if (apiKeys.length === 0) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
const fetchers = apiKeys.map(({key, rateLimitPerMinute}) => {
|
const fetchers = apiKeys.map(({key, rateLimitPerMinute}) => {
|
||||||
const bufferedRateLimit = applyRateLimitBuffer(rateLimitPerMinute);
|
const bufferedRateLimit = applyRateLimitBuffer(rateLimitPerMinute);
|
||||||
const interval = Math.ceil(60000 / bufferedRateLimit);
|
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 sheets = google.sheets({version: 'v4'});
|
||||||
const request = {spreadsheetId: documentId, ranges: sheetNames, key};
|
const request = {spreadsheetId: documentId, ranges: sheetNames, key};
|
||||||
const response = await sheets.spreadsheets.values.batchGet(request);
|
const response = await sheets.spreadsheets.values.batchGet(request);
|
||||||
@@ -126,15 +158,44 @@ function createApiKeyBatchFetcher(config) {
|
|||||||
rowsBySheet[sheetName] = valueRanges[i]?.values || [];
|
rowsBySheet[sheetName] = valueRanges[i]?.values || [];
|
||||||
}
|
}
|
||||||
return rowsBySheet;
|
return rowsBySheet;
|
||||||
});
|
};
|
||||||
return {key, fetch: throttled};
|
|
||||||
|
return {key, fetchRaw, interval, nextAvailable: 0};
|
||||||
});
|
});
|
||||||
|
|
||||||
let currentIndex = 0;
|
let currentIndex = 0;
|
||||||
|
|
||||||
|
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
|
||||||
|
|
||||||
return async (documentId, sheetNames) => {
|
return async (documentId, sheetNames) => {
|
||||||
const worker = fetchers[currentIndex];
|
const now = () => Date.now();
|
||||||
currentIndex = (currentIndex + 1) % fetchers.length;
|
|
||||||
return worker.fetch(documentId, sheetNames);
|
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);
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user