Spaces:
Sleeping
Sleeping
File size: 4,896 Bytes
548a458 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | // Per-session resource buffers for large stdout/stderr from bash_exec.
// Implements ring buffer over last N execs, byte caps per session and global.
export interface ResourceBuffer {
sessionId: string;
execId: string;
stream: "stdout" | "stderr";
bytes: Buffer;
complete: boolean; // true when not truncated below capturedOutputBytesExec cap
createdAt: number;
expiresAt: number;
}
export class ResourceStore {
private byKey = new Map<string, ResourceBuffer>(); // key: sessionId/execId/stream
private bySession = new Map<string, string[]>(); // sessionId -> [execId in insertion order]
private totalBytes = 0;
constructor(
private readonly maxExecsRetainedSession: number,
private readonly maxResourceBytesSession: number,
private readonly maxResourceBytesGlobal: number,
) {}
private key(sessionId: string, execId: string, stream: "stdout" | "stderr") {
return `${sessionId}::${execId}::${stream}`;
}
store(
sessionId: string,
execId: string,
stream: "stdout" | "stderr",
text: string,
complete: boolean,
ttlMs: number,
): boolean {
const bytes = Buffer.from(text, "utf8");
if (bytes.length === 0) return false;
if (bytes.length > this.maxResourceBytesGlobal) return false;
this.evictForCapacity(sessionId, bytes.length);
const buf: ResourceBuffer = {
sessionId,
execId,
stream,
bytes,
complete,
createdAt: Date.now(),
expiresAt: Date.now() + ttlMs,
};
this.byKey.set(this.key(sessionId, execId, stream), buf);
const list = this.bySession.get(sessionId) ?? [];
if (!list.includes(execId)) list.push(execId);
this.bySession.set(sessionId, list);
this.totalBytes += bytes.length;
return true;
}
get(
sessionId: string,
execId: string,
stream: "stdout" | "stderr",
): ResourceBuffer | null {
const b = this.byKey.get(this.key(sessionId, execId, stream));
if (!b) return null;
if (b.expiresAt < Date.now()) {
this.delete(sessionId, execId, stream);
return null;
}
return b;
}
delete(sessionId: string, execId: string, stream: "stdout" | "stderr") {
const k = this.key(sessionId, execId, stream);
const b = this.byKey.get(k);
if (b) {
this.totalBytes -= b.bytes.length;
this.byKey.delete(k);
}
}
deleteSession(sessionId: string) {
const execs = this.bySession.get(sessionId) ?? [];
for (const eid of execs) {
for (const s of ["stdout", "stderr"] as const) {
this.delete(sessionId, eid, s);
}
}
this.bySession.delete(sessionId);
}
totalBytesUsed() {
return this.totalBytes;
}
sessionBytesUsed(sessionId: string): number {
let total = 0;
const execs = this.bySession.get(sessionId) ?? [];
for (const eid of execs) {
for (const s of ["stdout", "stderr"] as const) {
const b = this.byKey.get(this.key(sessionId, eid, s));
if (b) total += b.bytes.length;
}
}
return total;
}
// Evict oldest exec resources for session until we fit incoming bytes within caps.
private evictForCapacity(sessionId: string, incomingBytes: number) {
// Per-session: if exec count limit would be exceeded, drop oldest. (Counts exec ids.)
const execs = this.bySession.get(sessionId) ?? [];
while (execs.length >= this.maxExecsRetainedSession) {
const oldest = execs.shift();
if (!oldest) break;
for (const s of ["stdout", "stderr"] as const) {
this.delete(sessionId, oldest, s);
}
}
// Per-session byte cap
while (this.sessionBytesUsed(sessionId) + incomingBytes > this.maxResourceBytesSession) {
const oldest = execs.shift();
if (!oldest) break;
for (const s of ["stdout", "stderr"] as const) {
this.delete(sessionId, oldest, s);
}
}
// Global byte cap: drop oldest entries across all sessions.
while (this.totalBytes + incomingBytes > this.maxResourceBytesGlobal) {
const allKeys = Array.from(this.byKey.values()).sort(
(a, b) => a.createdAt - b.createdAt,
);
const oldest = allKeys[0];
if (!oldest) break;
this.delete(oldest.sessionId, oldest.execId, oldest.stream);
// also prune from per-session list
const list = this.bySession.get(oldest.sessionId);
if (list) {
const idx = list.indexOf(oldest.execId);
const stillHas =
this.byKey.has(this.key(oldest.sessionId, oldest.execId, "stdout")) ||
this.byKey.has(this.key(oldest.sessionId, oldest.execId, "stderr"));
if (!stillHas && idx >= 0) list.splice(idx, 1);
}
}
this.bySession.set(sessionId, execs);
}
sweepExpired() {
const now = Date.now();
for (const [, b] of this.byKey) {
if (b.expiresAt < now) {
this.delete(b.sessionId, b.execId, b.stream);
}
}
}
}
|