Skip to content

feat: retry, concurrency control, and parallel DB optimizations #29

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
efenocchi merged 5 commits into from
Apr 10, 2026
Merged

feat: retry, concurrency control, and parallel DB optimizations #29

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
90db3f8
feat: add retry, concurrency control, and parallel optimizations to D…
khustup2 Apr 10, 2026
626662d
merge main, resolve conflicts with column renames (content_text → sum…
khustup2 Apr 10, 2026
5091a8d
feat: batch prefetch for grep — replace N+1 SQL queries with single S…
khustup2 Apr 10, 2026
b435530
fix: readFile now checks content cache, making prefetch effective for…
khustup2 Apr 10, 2026
c738ba9
remove bench/ from repo, add to .gitignore
khustup2 Apr 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,3 +6,4 @@ tmp/
.env
.env.*
coverage/
bench/
127 changes: 102 additions & 25 deletions claude-code/bundle/capture.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,39 +72,100 @@ function sqlStr(value) {

// dist/src/deeplake-api.js
var log2 = (msg) => log("sdk", msg);
var RETRYABLE_CODES = /* @__PURE__ */ new Set([429, 500, 502, 503, 504]);
var MAX_RETRIES = 3;
var BASE_DELAY_MS = 500;
var MAX_CONCURRENCY = 5;
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}
var Semaphore = class {
max;
waiting = [];
active = 0;
constructor(max) {
this.max = max;
}
async acquire() {
if (this.active < this.max) {
this.active++;
return;
}
await new Promise((resolve) => this.waiting.push(resolve));
}
release() {
this.active--;
const next = this.waiting.shift();
if (next) {
this.active++;
next();
}
}
};
var DeeplakeApi = class {
token;
apiUrl;
orgId;
workspaceId;
tableName;
_pendingRows = [];
_sem = new Semaphore(MAX_CONCURRENCY);
constructor(token, apiUrl, orgId, workspaceId, tableName) {
this.token = token;
this.apiUrl = apiUrl;
this.orgId = orgId;
this.workspaceId = workspaceId;
this.tableName = tableName;
}
/** Execute SQL and return results as row-objects. */
/** Execute SQL with retry on transient errors and bounded concurrency. */
async query(sql) {
const resp = await fetch(`${this.apiUrl}/workspaces/${this.workspaceId}/tables/query`, {
method: "POST",
headers: {
Authorization: `Bearer ${this.token}`,
"Content-Type": "application/json",
"X-Activeloop-Org-Id": this.orgId
},
body: JSON.stringify({ query: sql })
});
if (!resp.ok) {
await this._sem.acquire();
try {
return await this._queryWithRetry(sql);
} finally {
this._sem.release();
}
}
async _queryWithRetry(sql) {
let lastError;
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
let resp;
try {
resp = await fetch(`${this.apiUrl}/workspaces/${this.workspaceId}/tables/query`, {
method: "POST",
headers: {
Authorization: `Bearer ${this.token}`,
"Content-Type": "application/json",
"X-Activeloop-Org-Id": this.orgId
},
body: JSON.stringify({ query: sql })
});
} catch (e) {
lastError = e instanceof Error ? e : new Error(String(e));
if (attempt < MAX_RETRIES) {
const delay = BASE_DELAY_MS * Math.pow(2, attempt) + Math.random() * 200;
log2(`query retry ${attempt + 1}/${MAX_RETRIES} (fetch error: ${lastError.message}) in ${delay.toFixed(0)}ms`);
await sleep(delay);
continue;
}
throw lastError;
}
if (resp.ok) {
const raw = await resp.json();
if (!raw?.rows || !raw?.columns)
return [];
return raw.rows.map((row) => Object.fromEntries(raw.columns.map((col, i) => [col, row[i]])));
}
const text = await resp.text().catch(() => "");
if (attempt < MAX_RETRIES && RETRYABLE_CODES.has(resp.status)) {
const delay = BASE_DELAY_MS * Math.pow(2, attempt) + Math.random() * 200;
log2(`query retry ${attempt + 1}/${MAX_RETRIES} (${resp.status}) in ${delay.toFixed(0)}ms`);
await sleep(delay);
continue;
}
throw new Error(`Query failed: ${resp.status}: ${text.slice(0, 200)}`);
}
const raw = await resp.json();
if (!raw?.rows || !raw?.columns)
return [];
return raw.rows.map((row) => Object.fromEntries(raw.columns.map((col, i) => [col, row[i]])));
throw lastError ?? new Error("Query failed: max retries exceeded");
}
// ── Writes ──────────────────────────────────────────────────────────────────
/** Queue rows for writing. Call commit() to flush. */
Expand Down Expand Up @@ -162,18 +223,34 @@ var DeeplakeApi = class {
async createIndex(column) {
await this.query(`CREATE INDEX IF NOT EXISTS idx_${sqlStr(column)}_bm25 ON "${this.tableName}" USING deeplake_index ("${column}")`);
}
/** List all tables in the workspace. */
/** List all tables in the workspace (with retry). */
async listTables() {
const resp = await fetch(`${this.apiUrl}/workspaces/${this.workspaceId}/tables`, {
headers: {
Authorization: `Bearer ${this.token}`,
"X-Activeloop-Org-Id": this.orgId
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
try {
const resp = await fetch(`${this.apiUrl}/workspaces/${this.workspaceId}/tables`, {
headers: {
Authorization: `Bearer ${this.token}`,
"X-Activeloop-Org-Id": this.orgId
}
});
if (resp.ok) {
const data = await resp.json();
return (data.tables ?? []).map((t) => t.table_name);
}
if (attempt < MAX_RETRIES && RETRYABLE_CODES.has(resp.status)) {
await sleep(BASE_DELAY_MS * Math.pow(2, attempt) + Math.random() * 200);
continue;
}
return [];
} catch {
if (attempt < MAX_RETRIES) {
await sleep(BASE_DELAY_MS * Math.pow(2, attempt));
continue;
}
return [];
}
});
if (!resp.ok)
return [];
const data = await resp.json();
return (data.tables ?? []).map((t) => t.table_name);
}
return [];
}
/** Create the memory table if it doesn't already exist. Migrate columns on existing tables. */
async ensureTable(name) {
Expand Down
142 changes: 109 additions & 33 deletions claude-code/bundle/pre-tool-use.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,39 +80,100 @@ function sqlStr(value) {

// dist/src/deeplake-api.js
var log2 = (msg) => log("sdk", msg);
var RETRYABLE_CODES = /* @__PURE__ */ new Set([429, 500, 502, 503, 504]);
var MAX_RETRIES = 3;
var BASE_DELAY_MS = 500;
var MAX_CONCURRENCY = 5;
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}
var Semaphore = class {
max;
waiting = [];
active = 0;
constructor(max) {
this.max = max;
}
async acquire() {
if (this.active < this.max) {
this.active++;
return;
}
await new Promise((resolve) => this.waiting.push(resolve));
}
release() {
this.active--;
const next = this.waiting.shift();
if (next) {
this.active++;
next();
}
}
};
var DeeplakeApi = class {
token;
apiUrl;
orgId;
workspaceId;
tableName;
_pendingRows = [];
_sem = new Semaphore(MAX_CONCURRENCY);
constructor(token, apiUrl, orgId, workspaceId, tableName) {
this.token = token;
this.apiUrl = apiUrl;
this.orgId = orgId;
this.workspaceId = workspaceId;
this.tableName = tableName;
}
/** Execute SQL and return results as row-objects. */
/** Execute SQL with retry on transient errors and bounded concurrency. */
async query(sql) {
const resp = await fetch(`${this.apiUrl}/workspaces/${this.workspaceId}/tables/query`, {
method: "POST",
headers: {
Authorization: `Bearer ${this.token}`,
"Content-Type": "application/json",
"X-Activeloop-Org-Id": this.orgId
},
body: JSON.stringify({ query: sql })
});
if (!resp.ok) {
await this._sem.acquire();
try {
return await this._queryWithRetry(sql);
} finally {
this._sem.release();
}
}
async _queryWithRetry(sql) {
let lastError;
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
let resp;
try {
resp = await fetch(`${this.apiUrl}/workspaces/${this.workspaceId}/tables/query`, {
method: "POST",
headers: {
Authorization: `Bearer ${this.token}`,
"Content-Type": "application/json",
"X-Activeloop-Org-Id": this.orgId
},
body: JSON.stringify({ query: sql })
});
} catch (e) {
lastError = e instanceof Error ? e : new Error(String(e));
if (attempt < MAX_RETRIES) {
const delay = BASE_DELAY_MS * Math.pow(2, attempt) + Math.random() * 200;
log2(`query retry ${attempt + 1}/${MAX_RETRIES} (fetch error: ${lastError.message}) in ${delay.toFixed(0)}ms`);
await sleep(delay);
continue;
}
throw lastError;
}
if (resp.ok) {
const raw = await resp.json();
if (!raw?.rows || !raw?.columns)
return [];
return raw.rows.map((row) => Object.fromEntries(raw.columns.map((col, i) => [col, row[i]])));
}
const text = await resp.text().catch(() => "");
if (attempt < MAX_RETRIES && RETRYABLE_CODES.has(resp.status)) {
const delay = BASE_DELAY_MS * Math.pow(2, attempt) + Math.random() * 200;
log2(`query retry ${attempt + 1}/${MAX_RETRIES} (${resp.status}) in ${delay.toFixed(0)}ms`);
await sleep(delay);
continue;
}
throw new Error(`Query failed: ${resp.status}: ${text.slice(0, 200)}`);
}
const raw = await resp.json();
if (!raw?.rows || !raw?.columns)
return [];
return raw.rows.map((row) => Object.fromEntries(raw.columns.map((col, i) => [col, row[i]])));
throw lastError ?? new Error("Query failed: max retries exceeded");
}
// ── Writes ──────────────────────────────────────────────────────────────────
/** Queue rows for writing. Call commit() to flush. */
Expand Down Expand Up @@ -170,18 +231,34 @@ var DeeplakeApi = class {
async createIndex(column) {
await this.query(`CREATE INDEX IF NOT EXISTS idx_${sqlStr(column)}_bm25 ON "${this.tableName}" USING deeplake_index ("${column}")`);
}
/** List all tables in the workspace. */
/** List all tables in the workspace (with retry). */
async listTables() {
const resp = await fetch(`${this.apiUrl}/workspaces/${this.workspaceId}/tables`, {
headers: {
Authorization: `Bearer ${this.token}`,
"X-Activeloop-Org-Id": this.orgId
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
try {
const resp = await fetch(`${this.apiUrl}/workspaces/${this.workspaceId}/tables`, {
headers: {
Authorization: `Bearer ${this.token}`,
"X-Activeloop-Org-Id": this.orgId
}
});
if (resp.ok) {
const data = await resp.json();
return (data.tables ?? []).map((t) => t.table_name);
}
if (attempt < MAX_RETRIES && RETRYABLE_CODES.has(resp.status)) {
await sleep(BASE_DELAY_MS * Math.pow(2, attempt) + Math.random() * 200);
continue;
}
return [];
} catch {
if (attempt < MAX_RETRIES) {
await sleep(BASE_DELAY_MS * Math.pow(2, attempt));
continue;
}
return [];
}
});
if (!resp.ok)
return [];
const data = await resp.json();
return (data.tables ?? []).map((t) => t.table_name);
}
return [];
}
/** Create the memory table if it doesn't already exist. Migrate columns on existing tables. */
async ensureTable(name) {
Expand Down Expand Up @@ -462,16 +539,15 @@ async function main() {
const pattern = input.tool_input.pattern ?? "";
const ignoreCase = !!input.tool_input["-i"];
log3(`direct grep: ${pattern}`);
const pathRows = await api.query(`SELECT path FROM "${table}" WHERE summary ${ignoreCase ? "ILIKE" : "LIKE"} '%${sqlStr(pattern)}%' LIMIT 10`);
if (pathRows.length > 0) {
const rows = await api.query(`SELECT path, summary FROM "${table}" WHERE summary ${ignoreCase ? "ILIKE" : "LIKE"} '%${sqlStr(pattern)}%' LIMIT 5`);
if (rows.length > 0) {
const allResults = [];
for (const pr of pathRows.slice(0, 5)) {
const p = pr["path"];
const contentRows = await api.query(`SELECT summary FROM "${table}" WHERE path = '${sqlStr(p)}' LIMIT 1`);
if (!contentRows[0]?.["summary"])
const re = new RegExp(pattern.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"), ignoreCase ? "i" : "");
for (const row of rows) {
const p = row["path"];
const text = row["summary"];
if (!text)
continue;
const text = contentRows[0]["summary"];
const re = new RegExp(pattern.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"), ignoreCase ? "i" : "");
const matches = text.split("\n").filter((line) => re.test(line)).slice(0, 5).map((line) => `${p}:${line.slice(0, 300)}`);
allResults.push(...matches);
}
Expand Down
Loading
Loading