mirror of
https://github.com/garrytan/gbrain.git
synced 2026-08-14 17:02:19 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b0185306e1 |
@@ -21,14 +21,17 @@ GBrain is tuned for the Supabase **Transaction pooler** (port 6543): it
|
||||
auto-disables prepared statements there and routes `engine.transaction()`
|
||||
(migrations, DDL, sync imports) to a derived **direct** connection
|
||||
(`db.<ref>.supabase.co:5432`). That direct host is IPv6-only, so on an
|
||||
IPv4-only host, reads work but sync **silently skips most pages**. This is the
|
||||
number one cause of "sync ran but nothing happened."
|
||||
IPv4-only host it is unreachable. When that happens gbrain now falls back to
|
||||
the pooler automatically (one stderr warning, then single-pool mode for the
|
||||
rest of the process) — but the pooler's ~2-min statement timeout can truncate
|
||||
very long migrations or bulk imports.
|
||||
|
||||
Fix: make the direct connection reachable over IPv4. Either set
|
||||
`GBRAIN_DIRECT_DATABASE_URL` to the **Session pooler** string (port 5432 on the
|
||||
`pooler.supabase.com` host, IPv4), or enable Supabase's IPv4 add-on. Verify by
|
||||
running `gbrain sync` and checking that the page count in `gbrain stats` matches
|
||||
the syncable file count in the repo.
|
||||
`pooler.supabase.com` host, IPv4), or enable Supabase's IPv4 add-on.
|
||||
`GBRAIN_DISABLE_DIRECT_POOL=1` skips the direct pool (and the fallback warning)
|
||||
entirely. Verify by running `gbrain sync` and checking that the page count in
|
||||
`gbrain stats` matches the syncable file count in the repo.
|
||||
|
||||
### The Primitives
|
||||
|
||||
|
||||
+8
-5
@@ -2720,14 +2720,17 @@ GBrain is tuned for the Supabase **Transaction pooler** (port 6543): it
|
||||
auto-disables prepared statements there and routes `engine.transaction()`
|
||||
(migrations, DDL, sync imports) to a derived **direct** connection
|
||||
(`db.<ref>.supabase.co:5432`). That direct host is IPv6-only, so on an
|
||||
IPv4-only host, reads work but sync **silently skips most pages**. This is the
|
||||
number one cause of "sync ran but nothing happened."
|
||||
IPv4-only host it is unreachable. When that happens gbrain now falls back to
|
||||
the pooler automatically (one stderr warning, then single-pool mode for the
|
||||
rest of the process) — but the pooler's ~2-min statement timeout can truncate
|
||||
very long migrations or bulk imports.
|
||||
|
||||
Fix: make the direct connection reachable over IPv4. Either set
|
||||
`GBRAIN_DIRECT_DATABASE_URL` to the **Session pooler** string (port 5432 on the
|
||||
`pooler.supabase.com` host, IPv4), or enable Supabase's IPv4 add-on. Verify by
|
||||
running `gbrain sync` and checking that the page count in `gbrain stats` matches
|
||||
the syncable file count in the repo.
|
||||
`pooler.supabase.com` host, IPv4), or enable Supabase's IPv4 add-on.
|
||||
`GBRAIN_DISABLE_DIRECT_POOL=1` skips the direct pool (and the fallback warning)
|
||||
entirely. Verify by running `gbrain sync` and checking that the page count in
|
||||
`gbrain stats` matches the syncable file count in the repo.
|
||||
|
||||
### The Primitives
|
||||
|
||||
|
||||
+1
-1
@@ -2387,7 +2387,7 @@ JOBS (Minions)
|
||||
jobs get <id> Job details + history
|
||||
jobs cancel <id> Cancel job
|
||||
jobs retry <id> Re-queue failed/dead job
|
||||
jobs prune [--older-than 30d] [--status s,..] Clean old terminal jobs (0d = no age floor)
|
||||
jobs prune [--older-than 30d] Clean old jobs
|
||||
jobs stats Job health dashboard
|
||||
jobs work [--queue Q] Start worker daemon (Postgres only)
|
||||
|
||||
|
||||
@@ -1078,6 +1078,9 @@ async function initPostgres(opts: {
|
||||
console.warn(' Direct connections are IPv6 only and fail in many environments.');
|
||||
console.warn(' Use the Transaction pooler connection string instead (port 6543):');
|
||||
console.warn(' Supabase Dashboard > Connect (top bar) > Connection String > Transaction pooler');
|
||||
console.warn(' (With a pooler URL, gbrain derives a direct connection for DDL and falls back');
|
||||
console.warn(' to the pooler automatically if that host is unreachable. Power users:');
|
||||
console.warn(' GBRAIN_DIRECT_DATABASE_URL overrides the derived URL; GBRAIN_DISABLE_DIRECT_POOL=1 disables it.)');
|
||||
console.warn('');
|
||||
}
|
||||
|
||||
@@ -1091,6 +1094,9 @@ async function initPostgres(opts: {
|
||||
if (databaseUrl.includes('supabase.co') && (msg.includes('ECONNREFUSED') || msg.includes('ETIMEDOUT'))) {
|
||||
console.error('Connection failed. Supabase direct connections (db.*.supabase.co:5432) are IPv6 only.');
|
||||
console.error('Use the Transaction pooler connection string instead (port 6543).');
|
||||
console.error('(gbrain derives its own direct connection from pooler URLs for DDL; if that host is');
|
||||
console.error('unreachable it falls back to the pooler. GBRAIN_DIRECT_DATABASE_URL overrides the');
|
||||
console.error('derived URL; GBRAIN_DISABLE_DIRECT_POOL=1 disables the direct pool entirely.)');
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
|
||||
+5
-33
@@ -106,21 +106,6 @@ export function parseMaxRssFlag(args: string[]): number | undefined {
|
||||
return parsed;
|
||||
}
|
||||
|
||||
/** Terminal statuses `jobs prune --status` accepts (PR #2282). Matches what
|
||||
* queue.prune can safely delete; anything else (waiting/active/…) is live. */
|
||||
export const PRUNE_STATUSES = ['completed', 'failed', 'dead', 'cancelled'] as const satisfies readonly MinionJobStatus[];
|
||||
|
||||
/** Parse a `--status a,b,c` value into prune statuses. Throws on any value
|
||||
* outside PRUNE_STATUSES (fail-fast, mirrors parseNiceValue). */
|
||||
export function parsePruneStatuses(raw: string): MinionJobStatus[] {
|
||||
const requested = raw.split(',').map(s => s.trim()).filter(Boolean);
|
||||
const invalid = requested.filter(s => !(PRUNE_STATUSES as readonly string[]).includes(s));
|
||||
if (requested.length === 0 || invalid.length > 0) {
|
||||
throw new Error(`--status accepts a comma-separated subset of [${PRUNE_STATUSES.join(', ')}]${invalid.length ? `. Invalid: ${invalid.join(', ')}` : ''}`);
|
||||
}
|
||||
return requested as MinionJobStatus[];
|
||||
}
|
||||
|
||||
/** Parse `--nice N` (then `GBRAIN_NICE` env). Returns:
|
||||
* - undefined if absent (no priority change — inherit)
|
||||
* - the validated integer in [-20, 19] otherwise
|
||||
@@ -223,9 +208,7 @@ USAGE
|
||||
gbrain jobs get <id>
|
||||
gbrain jobs cancel <id>
|
||||
gbrain jobs retry <id>
|
||||
gbrain jobs prune [--older-than 30d] [--status completed,failed,dead,cancelled]
|
||||
(--older-than 0d = no age floor: deletes ALL
|
||||
matching terminal jobs; pair with --status)
|
||||
gbrain jobs prune [--older-than 30d]
|
||||
gbrain jobs delete <id>
|
||||
gbrain jobs stats
|
||||
gbrain jobs smoke
|
||||
@@ -617,27 +600,16 @@ HANDLER TYPES (built in)
|
||||
case 'prune': {
|
||||
const olderThanStr = parseFlag(args, '--older-than') ?? '30d';
|
||||
const days = parseInt(olderThanStr, 10);
|
||||
if (isNaN(days) || days < 0) {
|
||||
console.error('Error: --older-than must be a non-negative number (days). Example: --older-than 30d; --older-than 0d removes the age floor (deletes ALL matching terminal jobs).');
|
||||
if (isNaN(days) || days <= 0) {
|
||||
console.error('Error: --older-than must be a positive number (days). Example: --older-than 30d');
|
||||
process.exit(1);
|
||||
}
|
||||
const statusFlag = parseFlag(args, '--status');
|
||||
let statuses: MinionJobStatus[] | undefined;
|
||||
if (statusFlag !== undefined) {
|
||||
try { statuses = parsePruneStatuses(statusFlag); }
|
||||
catch (e) { console.error(`Error: ${e instanceof Error ? e.message : String(e)}`); process.exit(1); }
|
||||
}
|
||||
|
||||
try { await queue.ensureSchema(); }
|
||||
catch (e) { console.error(e instanceof Error ? e.message : String(e)); process.exit(1); }
|
||||
|
||||
const count = await queue.prune({
|
||||
olderThan: new Date(Date.now() - days * 86400000),
|
||||
...(statuses ? { status: statuses } : {}),
|
||||
});
|
||||
const statusLabel = statuses ? statuses.join('+') : 'completed+dead+cancelled';
|
||||
const ageLabel = days === 0 ? 'regardless of age' : `older than ${days} days`;
|
||||
console.log(`Pruned ${count} ${statusLabel} jobs ${ageLabel}.`);
|
||||
const count = await queue.prune({ olderThan: new Date(Date.now() - days * 86400000) });
|
||||
console.log(`Pruned ${count} jobs older than ${days} days.`);
|
||||
break;
|
||||
}
|
||||
|
||||
|
||||
@@ -167,6 +167,25 @@ export function deriveDirectUrl(url: string): string | null {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Error codes that mean "the direct host is unreachable from this network"
|
||||
* (#1641). The auto-derived db.<ref>.supabase.co host is IPv6-only without
|
||||
* the paid IPv4 add-on, so ENOTFOUND/ECONNREFUSED here is expected on
|
||||
* IPv4-only networks — we fall back to the pooler instead of failing init.
|
||||
*/
|
||||
const NETWORK_UNREACHABLE_CODES = [
|
||||
'ENOTFOUND', 'ECONNREFUSED', 'ENETUNREACH', 'EHOSTUNREACH',
|
||||
'ETIMEDOUT', 'CONNECT_TIMEOUT',
|
||||
];
|
||||
|
||||
/** True when err looks like a network-unreachable failure (not auth/SQL). */
|
||||
export function isNetworkUnreachableError(err: unknown): boolean {
|
||||
const code = (err as { code?: unknown } | null)?.code;
|
||||
if (typeof code === 'string' && NETWORK_UNREACHABLE_CODES.includes(code)) return true;
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
return NETWORK_UNREACHABLE_CODES.some(c => msg.includes(c));
|
||||
}
|
||||
|
||||
/**
|
||||
* Read kill-switch state from env. Subordinate to parent manager's state
|
||||
* when present (A2 inheritance).
|
||||
@@ -319,7 +338,30 @@ export class ConnectionManager {
|
||||
throw err;
|
||||
});
|
||||
}
|
||||
const pool = await this._directInit;
|
||||
let pool: Sql | null;
|
||||
try {
|
||||
pool = await this._directInit;
|
||||
} catch (err) {
|
||||
// #1641: the derived direct host (db.<ref>.supabase.co) is IPv6-only
|
||||
// without Supabase's IPv4 add-on. On IPv4-only networks the direct
|
||||
// pool can never connect — permanently fall back to the read pool
|
||||
// (self-activating kill-switch) instead of failing init/migrations.
|
||||
// Non-network errors (auth, SQL) still throw: they mean misconfig,
|
||||
// not unreachability.
|
||||
if (isNetworkUnreachableError(err)) {
|
||||
const alreadyWarned = this._killSwitch;
|
||||
this._killSwitch = true;
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
if (!alreadyWarned) console.error(
|
||||
`gbrain: direct connection to ${this._directUrl ? this.hostOnly(this._directUrl) : 'unknown host'} unreachable (${msg}); ` +
|
||||
'falling back to the pooler for DDL/bulk (long migrations may hit the pooler statement timeout). ' +
|
||||
'Set GBRAIN_DIRECT_DATABASE_URL to a reachable direct URL (e.g. the Session pooler, port 5432) or enable the Supabase IPv4 add-on; ' +
|
||||
'GBRAIN_DISABLE_DIRECT_POOL=1 silences this.',
|
||||
);
|
||||
return this.getReadPool();
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
if (!pool) {
|
||||
// Defensive — initDirectPool should have thrown.
|
||||
throw new Error('connection-manager: direct pool init returned null');
|
||||
@@ -350,8 +392,9 @@ export class ConnectionManager {
|
||||
},
|
||||
};
|
||||
const t0 = Date.now();
|
||||
let pool: Sql | null = null;
|
||||
try {
|
||||
const pool = postgres(this._directUrl, opts);
|
||||
pool = postgres(this._directUrl, opts);
|
||||
// Probe to validate connectivity early.
|
||||
await pool`SELECT 1`;
|
||||
logConnectionEvent({
|
||||
@@ -362,6 +405,9 @@ export class ConnectionManager {
|
||||
});
|
||||
return pool;
|
||||
} catch (err) {
|
||||
// Don't leak the failed pool's sockets/timers (#1641 fallback keeps
|
||||
// the process running afterward).
|
||||
if (pool) await endPoolBounded(pool);
|
||||
logConnectionEvent({
|
||||
pool: 'ddl',
|
||||
op: 'error',
|
||||
|
||||
@@ -3,6 +3,7 @@ import {
|
||||
isSupabasePoolerUrl,
|
||||
deriveDirectUrl,
|
||||
readKillSwitchEnv,
|
||||
isNetworkUnreachableError,
|
||||
resolveDirectPoolSize,
|
||||
ConnectionManager,
|
||||
DEFAULT_DIRECT_POOL_SIZE,
|
||||
@@ -238,3 +239,65 @@ describe('ConnectionManager — parent inheritance (A2)', () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('isNetworkUnreachableError (#1641)', () => {
|
||||
test('classifies network codes as unreachable', () => {
|
||||
for (const code of ['ENOTFOUND', 'ECONNREFUSED', 'ENETUNREACH', 'EHOSTUNREACH', 'ETIMEDOUT', 'CONNECT_TIMEOUT']) {
|
||||
const err = Object.assign(new Error('connect failed'), { code });
|
||||
expect(isNetworkUnreachableError(err)).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
test('classifies by message when code absent', () => {
|
||||
expect(isNetworkUnreachableError(new Error('getaddrinfo ENOTFOUND db.abc.supabase.co'))).toBe(true);
|
||||
});
|
||||
|
||||
test('auth/SQL errors are NOT unreachable', () => {
|
||||
expect(isNetworkUnreachableError(new Error('password authentication failed for user "postgres"'))).toBe(false);
|
||||
expect(isNetworkUnreachableError(new Error('syntax error at or near "SELEC"'))).toBe(false);
|
||||
expect(isNetworkUnreachableError(null)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('ConnectionManager — direct-pool fallback on unreachable host (#1641)', () => {
|
||||
let originalKillSwitch: string | undefined;
|
||||
let originalError: typeof console.error;
|
||||
let errLines: string[];
|
||||
beforeEach(() => {
|
||||
originalKillSwitch = process.env.GBRAIN_DISABLE_DIRECT_POOL;
|
||||
delete process.env.GBRAIN_DISABLE_DIRECT_POOL;
|
||||
originalError = console.error;
|
||||
errLines = [];
|
||||
console.error = (...args: unknown[]) => { errLines.push(args.join(' ')); };
|
||||
});
|
||||
afterEach(() => {
|
||||
console.error = originalError;
|
||||
if (originalKillSwitch === undefined) delete process.env.GBRAIN_DISABLE_DIRECT_POOL;
|
||||
else process.env.GBRAIN_DISABLE_DIRECT_POOL = originalKillSwitch;
|
||||
});
|
||||
|
||||
test('ddl() falls back to the read pool when the direct host is unreachable', async () => {
|
||||
const cm = new ConnectionManager({
|
||||
url: 'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db',
|
||||
// 127.0.0.1:9 (discard) → instant ECONNREFUSED, the IPv4-only-network shape.
|
||||
directUrl: 'postgresql://postgres:p@127.0.0.1:9/db',
|
||||
});
|
||||
const fakeReadPool = {} as ReturnType<typeof ConnectionManager.prototype.read>;
|
||||
cm.setReadPool(fakeReadPool);
|
||||
expect(cm.isDualPoolActive()).toBe(true);
|
||||
|
||||
const pool = await cm.ddl(); // without the fix this throws ECONNREFUSED
|
||||
expect(pool).toBe(fakeReadPool);
|
||||
// Self-activating kill-switch: subsequent calls skip the direct pool.
|
||||
expect(cm.isKillSwitchActive()).toBe(true);
|
||||
expect(cm.isDualPoolActive()).toBe(false);
|
||||
expect(cm.describeMode().mode).toBe('single (kill-switch)');
|
||||
// One stderr line mentioning the power-user override.
|
||||
const warning = errLines.filter(l => l.includes('GBRAIN_DIRECT_DATABASE_URL'));
|
||||
expect(warning.length).toBe(1);
|
||||
|
||||
const again = await cm.ddl();
|
||||
expect(again).toBe(fakeReadPool);
|
||||
expect(errLines.filter(l => l.includes('GBRAIN_DIRECT_DATABASE_URL')).length).toBe(1);
|
||||
}, 20000);
|
||||
});
|
||||
|
||||
@@ -1,30 +0,0 @@
|
||||
/**
|
||||
* Unit tests for parsePruneStatuses (PR #2282) — `jobs prune --status` parsing.
|
||||
*/
|
||||
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { parsePruneStatuses, PRUNE_STATUSES } from '../src/commands/jobs.ts';
|
||||
|
||||
describe('parsePruneStatuses', () => {
|
||||
test('parses a single status', () => {
|
||||
expect(parsePruneStatuses('failed')).toEqual(['failed']);
|
||||
});
|
||||
|
||||
test('parses a comma-separated list with whitespace', () => {
|
||||
expect(parsePruneStatuses(' completed, dead ')).toEqual(['completed', 'dead']);
|
||||
});
|
||||
|
||||
test('accepts every documented terminal status', () => {
|
||||
expect(parsePruneStatuses(PRUNE_STATUSES.join(','))).toEqual([...PRUNE_STATUSES]);
|
||||
});
|
||||
|
||||
test('throws on non-terminal statuses', () => {
|
||||
expect(() => parsePruneStatuses('waiting')).toThrow(/Invalid: waiting/);
|
||||
expect(() => parsePruneStatuses('completed,active')).toThrow(/Invalid: active/);
|
||||
});
|
||||
|
||||
test('throws on empty value', () => {
|
||||
expect(() => parsePruneStatuses('')).toThrow(/comma-separated subset/);
|
||||
expect(() => parsePruneStatuses(',')).toThrow(/comma-separated subset/);
|
||||
});
|
||||
});
|
||||
@@ -702,32 +702,6 @@ describe('MinionQueue: Prune', () => {
|
||||
const count = await queue.prune({ olderThan: new Date(Date.now() + 86400000) }); // future date = prune everything old enough
|
||||
expect(count).toBe(1); // only the cancelled one
|
||||
});
|
||||
|
||||
// PR #2282: `jobs prune --status` passes an explicit status subset through.
|
||||
test('status filter prunes only the requested terminal statuses', async () => {
|
||||
const cancelled = await queue.add('sync', {});
|
||||
await queue.cancelJob(cancelled.id);
|
||||
const dead = await queue.add('embed', {}, { max_attempts: 1 });
|
||||
await queue.claim('tok1', 30000, 'default', ['embed']);
|
||||
await queue.failJob(dead.id, 'tok1', 'boom', 'dead');
|
||||
|
||||
const count = await queue.prune({ olderThan: new Date(Date.now() + 86400000), status: ['dead'] });
|
||||
expect(count).toBe(1); // only the dead one
|
||||
|
||||
const remaining = await queue.getJobs({ status: 'cancelled' });
|
||||
expect(remaining.length).toBe(1);
|
||||
});
|
||||
|
||||
// PR #2282: `--older-than 0d` = no age floor — olderThan of "now" deletes
|
||||
// terminal jobs that finished moments ago.
|
||||
test('olderThan now (0d semantics) prunes just-terminated jobs', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.cancelJob(job.id);
|
||||
await new Promise(r => setTimeout(r, 5)); // ensure updated_at < now
|
||||
|
||||
const count = await queue.prune({ olderThan: new Date() });
|
||||
expect(count).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
// --- Stats (1 test) ---
|
||||
|
||||
Reference in New Issue
Block a user