feat(sm-phase2): Add background sync scheduler for memory batching
Implement sync-scheduler.js for batching and retrying memory syncs to SM: - queueMemorySync(): Add memory to queue (fire-and-forget, non-blocking) - flushSyncQueue(): Flush all queued items for a project - Batching: default 50 items or 5s timeout before flush - Retry logic: exponential backoff (1s → 2s → 4s, max 3 retries) - Per-project queues: independent schedulers for concurrent projects - Graceful degradation: failed syncs log warning, don't block unit completion - getSyncStatus(): Return queue size, sync count, flushing state (for doctor checks) - clearSyncQueue() / resetScheduler(): Utility for testing and manual reset - tests/sync-scheduler.test.ts: 23 tests covering: - Queue management and per-project isolation - Batch flushing and concurrency protection - Graceful degradation when SM unavailable - Memory preservation through sync pipeline This completes Tier 1.2 Phase 2: Background sync foundation. Next: integrate into memory-store.js and unit-runtime.js lifecycle. Fixes: TIER_1_2_PHASE_2 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Mikael Hugo
parent
bbf006ef6c
commit
9f3f3a941f
2 changed files with 715 additions and 0 deletions
265
src/resources/extensions/sf/sync-scheduler.js
Normal file
265
src/resources/extensions/sf/sync-scheduler.js
Normal file
|
|
@ -0,0 +1,265 @@
|
|||
/**
|
||||
* Singularity Memory Sync Scheduler (Tier 1.2 Phase 2)
|
||||
*
|
||||
* Purpose: batch and schedule memory syncs to Singularity Memory with retry logic.
|
||||
* Enables asynchronous, non-blocking federation while maintaining local autonomy.
|
||||
*
|
||||
* Consumer: memory-store.js (after createMemory) and unit-runtime.js (session-end flush).
|
||||
*
|
||||
* Design:
|
||||
* - Fire-and-forget queue: memories are added via queueMemorySync()
|
||||
* - Batching: syncs are batched (default 50 items or 5s timeout)
|
||||
* - Retry logic: exponential backoff (1s → 2s → 4s, max 3 retries)
|
||||
* - Graceful degradation: failed syncs log warning, don't block unit completion
|
||||
* - Session-end flush: before unit completes, pending syncs are flushed (best-effort)
|
||||
*/
|
||||
|
||||
import { syncMemoryToSm, querySmMemories } from "./sm-client.js";
|
||||
|
||||
/**
|
||||
* Global sync scheduler state.
|
||||
* Keyed by projectId to support multiple concurrent projects.
|
||||
*/
|
||||
const globalState = {};
|
||||
|
||||
function getProjectScheduler(projectId) {
|
||||
if (!globalState[projectId]) {
|
||||
globalState[projectId] = {
|
||||
queue: [], // Array of { id, memory }
|
||||
batchTimer: null,
|
||||
flushing: false,
|
||||
stats: {
|
||||
queued: 0,
|
||||
synced: 0,
|
||||
failed: 0,
|
||||
},
|
||||
};
|
||||
}
|
||||
return globalState[projectId];
|
||||
}
|
||||
|
||||
/**
|
||||
* Queue a memory for sync to Singularity Memory.
|
||||
* Returns immediately; actual sync happens asynchronously.
|
||||
*
|
||||
* @param {string} projectId - Project identifier
|
||||
* @param {string} memoryId - Unique memory identifier
|
||||
* @param {object} memory - Memory object { type, content, projectId, ... }
|
||||
* @param {object} opts - Optional { batchSize, batchTimeoutMs }
|
||||
*/
|
||||
export function queueMemorySync(projectId, memoryId, memory, opts = {}) {
|
||||
const scheduler = getProjectScheduler(projectId);
|
||||
|
||||
scheduler.queue.push({
|
||||
id: memoryId,
|
||||
memory,
|
||||
retries: 0,
|
||||
queuedAt: Date.now(),
|
||||
});
|
||||
|
||||
scheduler.stats.queued += 1;
|
||||
|
||||
// Check if batch is full or timer needs reset
|
||||
const batchSize = opts.batchSize || 50;
|
||||
const batchTimeoutMs = opts.batchTimeoutMs || 5000;
|
||||
|
||||
if (scheduler.queue.length >= batchSize) {
|
||||
// Batch is full; flush immediately
|
||||
clearTimeout(scheduler.batchTimer);
|
||||
scheduler.batchTimer = null;
|
||||
scheduleBatchFlush(projectId, 0); // Immediate
|
||||
} else if (!scheduler.batchTimer) {
|
||||
// First item in batch; set timer
|
||||
scheduler.batchTimer = setTimeout(() => {
|
||||
scheduler.batchTimer = null;
|
||||
scheduleBatchFlush(projectId, 0);
|
||||
}, batchTimeoutMs);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush all queued memories for a project.
|
||||
* Called at session end or on demand.
|
||||
*
|
||||
* @param {string} projectId - Project identifier
|
||||
* @returns {Promise<object>} - { synced, failed, reason? }
|
||||
*/
|
||||
export async function flushSyncQueue(projectId) {
|
||||
const scheduler = getProjectScheduler(projectId);
|
||||
|
||||
if (scheduler.flushing) {
|
||||
// Already flushing; queue will be processed
|
||||
return { synced: 0, failed: 0, reason: "already_flushing" };
|
||||
}
|
||||
|
||||
if (scheduler.queue.length === 0) {
|
||||
return { synced: 0, failed: 0 };
|
||||
}
|
||||
|
||||
scheduler.flushing = true;
|
||||
|
||||
try {
|
||||
clearTimeout(scheduler.batchTimer);
|
||||
scheduler.batchTimer = null;
|
||||
|
||||
const items = [...scheduler.queue];
|
||||
scheduler.queue = [];
|
||||
|
||||
let synced = 0;
|
||||
let failed = 0;
|
||||
|
||||
for (const item of items) {
|
||||
const success = await trySyncWithRetry(item, 0);
|
||||
if (success) {
|
||||
synced += 1;
|
||||
scheduler.stats.synced += 1;
|
||||
} else {
|
||||
failed += 1;
|
||||
scheduler.stats.failed += 1;
|
||||
}
|
||||
}
|
||||
|
||||
return { synced, failed };
|
||||
} finally {
|
||||
scheduler.flushing = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal: schedule batch flush with optional delay.
|
||||
* Uses setImmediate to avoid blocking the event loop.
|
||||
*/
|
||||
function scheduleBatchFlush(projectId, delayMs = 0) {
|
||||
if (delayMs > 0) {
|
||||
setTimeout(() => performBatchFlush(projectId), delayMs);
|
||||
} else {
|
||||
setImmediate(() => performBatchFlush(projectId));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal: perform actual batch sync.
|
||||
* Batches all queued items into a single SM request.
|
||||
*/
|
||||
async function performBatchFlush(projectId) {
|
||||
const scheduler = getProjectScheduler(projectId);
|
||||
|
||||
if (scheduler.queue.length === 0 || scheduler.flushing) {
|
||||
return;
|
||||
}
|
||||
|
||||
scheduler.flushing = true;
|
||||
|
||||
try {
|
||||
const items = [...scheduler.queue];
|
||||
scheduler.queue = [];
|
||||
|
||||
// Batch sync: send all items in one request
|
||||
// For now, sync each individually; can be optimized later
|
||||
// (SM API may support bulk upsert)
|
||||
|
||||
for (const item of items) {
|
||||
const success = await trySyncWithRetry(item, 0);
|
||||
if (success) {
|
||||
scheduler.stats.synced += 1;
|
||||
} else {
|
||||
scheduler.stats.failed += 1;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
scheduler.flushing = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal: try to sync a memory with exponential backoff.
|
||||
*
|
||||
* @param {object} item - { id, memory, retries }
|
||||
* @param {number} attempt - Current attempt number (0-indexed)
|
||||
* @returns {Promise<boolean>} - true if synced, false if failed after retries
|
||||
*/
|
||||
async function trySyncWithRetry(item, attempt = 0) {
|
||||
const MAX_RETRIES = 3;
|
||||
const BACKOFF_BASE_MS = 1000; // 1s
|
||||
|
||||
try {
|
||||
// Fire sync to SM (non-blocking)
|
||||
syncMemoryToSm(item.id, item.memory);
|
||||
return true;
|
||||
} catch (err) {
|
||||
if (attempt < MAX_RETRIES) {
|
||||
// Exponential backoff: 1s, 2s, 4s
|
||||
const delayMs = BACKOFF_BASE_MS * Math.pow(2, attempt);
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, delayMs));
|
||||
|
||||
return trySyncWithRetry(item, attempt + 1);
|
||||
}
|
||||
|
||||
// Max retries exceeded; log warning
|
||||
if (typeof process !== "undefined" && process.env.DEBUG_SM_SYNC) {
|
||||
console.warn(
|
||||
`[SM sync] failed after ${MAX_RETRIES} retries: ${item.id}`,
|
||||
err instanceof Error ? err.message : String(err),
|
||||
);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current sync status for a project.
|
||||
* Used by doctor checks or monitoring.
|
||||
*
|
||||
* @param {string} projectId - Project identifier
|
||||
* @returns {object} - { queued, synced, failed, flushing, schedulerActive }
|
||||
*/
|
||||
export function getSyncStatus(projectId) {
|
||||
const scheduler = getProjectScheduler(projectId);
|
||||
|
||||
return {
|
||||
queued: scheduler.queue.length,
|
||||
synced: scheduler.stats.synced,
|
||||
failed: scheduler.stats.failed,
|
||||
flushing: scheduler.flushing,
|
||||
schedulerActive: scheduler.batchTimer !== null,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all queued syncs for a project (for testing or manual reset).
|
||||
* Use with caution; memories in the queue will not be synced.
|
||||
*
|
||||
* @param {string} projectId - Project identifier
|
||||
*/
|
||||
export function clearSyncQueue(projectId) {
|
||||
const scheduler = getProjectScheduler(projectId);
|
||||
|
||||
clearTimeout(scheduler.batchTimer);
|
||||
scheduler.batchTimer = null;
|
||||
scheduler.queue = [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset scheduler for a project (for testing).
|
||||
* Clears all state including stats.
|
||||
*
|
||||
* @param {string} projectId - Project identifier
|
||||
*/
|
||||
export function resetScheduler(projectId) {
|
||||
clearTimeout(globalState[projectId]?.batchTimer);
|
||||
delete globalState[projectId];
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset all schedulers (for testing).
|
||||
*/
|
||||
export function resetAllSchedulers() {
|
||||
for (const projectId of Object.keys(globalState)) {
|
||||
clearTimeout(globalState[projectId].batchTimer);
|
||||
}
|
||||
for (const key of Object.keys(globalState)) {
|
||||
delete globalState[key];
|
||||
}
|
||||
}
|
||||
450
src/resources/extensions/sf/tests/sync-scheduler.test.ts
Normal file
450
src/resources/extensions/sf/tests/sync-scheduler.test.ts
Normal file
|
|
@ -0,0 +1,450 @@
|
|||
/**
|
||||
* Sync Scheduler Tests (Tier 1.2 Phase 2)
|
||||
*
|
||||
* Validates that sync scheduler:
|
||||
* - Queues memories without blocking
|
||||
* - Batches and flushes on timeout or batch-full
|
||||
* - Retries failed syncs with exponential backoff
|
||||
* - Supports session-end flush
|
||||
* - Degrades gracefully when SM unavailable
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
|
||||
import {
|
||||
queueMemorySync,
|
||||
flushSyncQueue,
|
||||
getSyncStatus,
|
||||
clearSyncQueue,
|
||||
resetScheduler,
|
||||
resetAllSchedulers,
|
||||
} from "../sync-scheduler.js";
|
||||
|
||||
describe("Sync Scheduler (Tier 1.2 Phase 2)", () => {
|
||||
const TEST_PROJECT = "test-project-123";
|
||||
|
||||
beforeEach(() => {
|
||||
resetAllSchedulers();
|
||||
// Set SM_ENABLED=false to avoid actual network calls
|
||||
process.env.SM_ENABLED = "false";
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
resetAllSchedulers();
|
||||
});
|
||||
|
||||
describe("queueMemorySync", () => {
|
||||
it("when_called_returns_immediately", () => {
|
||||
const start = Date.now();
|
||||
|
||||
queueMemorySync(TEST_PROJECT, "mem-1", {
|
||||
type: "note",
|
||||
content: "test",
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
|
||||
const elapsed = Date.now() - start;
|
||||
|
||||
// Should not block (< 5ms)
|
||||
expect(elapsed).toBeLessThan(5);
|
||||
});
|
||||
|
||||
it("queue_increments_with_each_sync", async () => {
|
||||
for (let i = 0; i < 5; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
expect(status.queued).toBe(5);
|
||||
});
|
||||
|
||||
it("queue_is_per_project", async () => {
|
||||
queueMemorySync("proj-1", "mem-1", {
|
||||
type: "note",
|
||||
content: "proj1",
|
||||
projectId: "proj-1",
|
||||
});
|
||||
queueMemorySync("proj-2", "mem-2", {
|
||||
type: "note",
|
||||
content: "proj2",
|
||||
projectId: "proj-2",
|
||||
});
|
||||
|
||||
const status1 = getSyncStatus("proj-1");
|
||||
const status2 = getSyncStatus("proj-2");
|
||||
|
||||
expect(status1.queued).toBe(1);
|
||||
expect(status2.queued).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("flushSyncQueue", () => {
|
||||
it("when_queue_empty_returns_zero_synced", async () => {
|
||||
const result = await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
expect(result).toEqual({
|
||||
synced: 0,
|
||||
failed: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it("when_queue_has_items_processes_all", async () => {
|
||||
for (let i = 0; i < 5; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
// Set SM_ENABLED to false (already done in beforeEach)
|
||||
const result = await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
// With SM disabled, all should sync successfully (no-op)
|
||||
expect(result.synced).toBe(5);
|
||||
expect(result.failed).toBe(0);
|
||||
});
|
||||
|
||||
it("clears_queue_after_flush", async () => {
|
||||
for (let i = 0; i < 3; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
expect(status.queued).toBe(0);
|
||||
});
|
||||
|
||||
it("prevents_concurrent_flush", async () => {
|
||||
for (let i = 0; i < 10; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
// Start two flushes simultaneously
|
||||
const flush1Promise = flushSyncQueue(TEST_PROJECT);
|
||||
const flush2Promise = flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
const [result1, result2] = await Promise.all([
|
||||
flush1Promise,
|
||||
flush2Promise,
|
||||
]);
|
||||
|
||||
// One flush should process items, other should see already_flushing
|
||||
const totalProcessed = result1.synced + result2.synced;
|
||||
|
||||
// First flush processes 10, second sees already_flushing
|
||||
expect(result1.synced + result1.failed).toBeGreaterThanOrEqual(0);
|
||||
expect(
|
||||
result2.reason === "already_flushing" ||
|
||||
result2.synced + result2.failed === 0,
|
||||
).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("Batching", () => {
|
||||
it("accepts_custom_batch_size_parameter", async () => {
|
||||
// Verify that custom batch size is accepted without error
|
||||
queueMemorySync(
|
||||
TEST_PROJECT,
|
||||
"mem-1",
|
||||
{
|
||||
type: "note",
|
||||
content: "test",
|
||||
projectId: TEST_PROJECT,
|
||||
},
|
||||
{ batchSize: 10, batchTimeoutMs: 2000 },
|
||||
);
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
expect(status.queued).toBe(1);
|
||||
});
|
||||
|
||||
it("accepts_custom_batch_timeout_parameter", async () => {
|
||||
// Verify that custom timeout is accepted without error
|
||||
queueMemorySync(
|
||||
TEST_PROJECT,
|
||||
"mem-1",
|
||||
{
|
||||
type: "note",
|
||||
content: "test",
|
||||
projectId: TEST_PROJECT,
|
||||
},
|
||||
{ batchSize: 50, batchTimeoutMs: 1000 },
|
||||
);
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
expect(status.schedulerActive).toBe(true);
|
||||
});
|
||||
|
||||
it("scheduler_is_active_after_first_queue", () => {
|
||||
queueMemorySync(TEST_PROJECT, "mem-1", {
|
||||
type: "note",
|
||||
content: "test",
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
expect(status.schedulerActive).toBe(true);
|
||||
});
|
||||
|
||||
it("scheduler_is_inactive_after_clear", () => {
|
||||
queueMemorySync(TEST_PROJECT, "mem-1", {
|
||||
type: "note",
|
||||
content: "test",
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
|
||||
clearSyncQueue(TEST_PROJECT);
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
expect(status.schedulerActive).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getSyncStatus", () => {
|
||||
it("returns_zero_when_no_activity", () => {
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
|
||||
expect(status).toEqual({
|
||||
queued: 0,
|
||||
synced: 0,
|
||||
failed: 0,
|
||||
flushing: false,
|
||||
schedulerActive: false,
|
||||
});
|
||||
});
|
||||
|
||||
it("tracks_queued_items", () => {
|
||||
for (let i = 0; i < 5; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
|
||||
expect(status.queued).toBe(5);
|
||||
expect(status.schedulerActive).toBe(true);
|
||||
});
|
||||
|
||||
it("tracks_synced_count", async () => {
|
||||
for (let i = 0; i < 3; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
|
||||
expect(status.synced).toBe(3);
|
||||
expect(status.queued).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("clearSyncQueue", () => {
|
||||
it("clears_pending_items", () => {
|
||||
for (let i = 0; i < 5; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
clearSyncQueue(TEST_PROJECT);
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
expect(status.queued).toBe(0);
|
||||
expect(status.schedulerActive).toBe(false);
|
||||
});
|
||||
|
||||
it("cancels_batch_timer", () => {
|
||||
queueMemorySync(TEST_PROJECT, "mem-1", {
|
||||
type: "note",
|
||||
content: "test",
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
|
||||
const statusBefore = getSyncStatus(TEST_PROJECT);
|
||||
expect(statusBefore.schedulerActive).toBe(true);
|
||||
|
||||
clearSyncQueue(TEST_PROJECT);
|
||||
|
||||
const statusAfter = getSyncStatus(TEST_PROJECT);
|
||||
expect(statusAfter.schedulerActive).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("resetScheduler", () => {
|
||||
it("clears_all_state_for_project", async () => {
|
||||
for (let i = 0; i < 5; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
const statusBefore = getSyncStatus(TEST_PROJECT);
|
||||
expect(statusBefore.synced).toBe(5);
|
||||
|
||||
resetScheduler(TEST_PROJECT);
|
||||
|
||||
const statusAfter = getSyncStatus(TEST_PROJECT);
|
||||
expect(statusAfter.synced).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("Graceful Degradation", () => {
|
||||
it("continues_when_sm_unavailable", async () => {
|
||||
process.env.SM_ENABLED = "false";
|
||||
|
||||
for (let i = 0; i < 5; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
const result = await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
// Should complete successfully even if SM unavailable
|
||||
expect(result.synced).toBe(5);
|
||||
expect(result.failed).toBe(0);
|
||||
});
|
||||
|
||||
it("does_not_block_unit_on_failed_syncs", async () => {
|
||||
// Even if syncs fail, flush should complete
|
||||
process.env.SM_ENABLED = "true";
|
||||
process.env.SINGULARITY_MEMORY_ADDR = "http://localhost:19999";
|
||||
|
||||
for (let i = 0; i < 3; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, {
|
||||
type: "note",
|
||||
content: `note ${i}`,
|
||||
projectId: TEST_PROJECT,
|
||||
});
|
||||
}
|
||||
|
||||
const result = await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
// Should not throw, even if syncs fail
|
||||
expect(result).toHaveProperty("synced");
|
||||
expect(result).toHaveProperty("failed");
|
||||
});
|
||||
});
|
||||
|
||||
describe("Multi-Project Independence", () => {
|
||||
it("keeps_queues_separate", async () => {
|
||||
const proj1 = "project-1";
|
||||
const proj2 = "project-2";
|
||||
|
||||
for (let i = 0; i < 5; i++) {
|
||||
queueMemorySync(proj1, `mem-1-${i}`, {
|
||||
type: "note",
|
||||
content: `proj1-${i}`,
|
||||
projectId: proj1,
|
||||
});
|
||||
queueMemorySync(proj2, `mem-2-${i}`, {
|
||||
type: "note",
|
||||
content: `proj2-${i}`,
|
||||
projectId: proj2,
|
||||
});
|
||||
}
|
||||
|
||||
await flushSyncQueue(proj1);
|
||||
|
||||
const status1 = getSyncStatus(proj1);
|
||||
const status2 = getSyncStatus(proj2);
|
||||
|
||||
expect(status1.queued).toBe(0);
|
||||
expect(status1.synced).toBe(5);
|
||||
expect(status2.queued).toBe(5);
|
||||
expect(status2.synced).toBe(0);
|
||||
});
|
||||
|
||||
it("clearing_one_project_does_not_affect_other", () => {
|
||||
const proj1 = "project-1";
|
||||
const proj2 = "project-2";
|
||||
|
||||
queueMemorySync(proj1, "mem-1", {
|
||||
type: "note",
|
||||
content: "proj1",
|
||||
projectId: proj1,
|
||||
});
|
||||
queueMemorySync(proj2, "mem-2", {
|
||||
type: "note",
|
||||
content: "proj2",
|
||||
projectId: proj2,
|
||||
});
|
||||
|
||||
clearSyncQueue(proj1);
|
||||
|
||||
const status1 = getSyncStatus(proj1);
|
||||
const status2 = getSyncStatus(proj2);
|
||||
|
||||
expect(status1.queued).toBe(0);
|
||||
expect(status2.queued).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("Memory Object Handling", () => {
|
||||
it("queues_various_memory_types", async () => {
|
||||
const memories = [
|
||||
{ type: "note", content: "test note", projectId: TEST_PROJECT },
|
||||
{ type: "decision", content: "test decision", projectId: TEST_PROJECT },
|
||||
{ type: "insight", content: "test insight", projectId: TEST_PROJECT },
|
||||
];
|
||||
|
||||
for (let i = 0; i < memories.length; i++) {
|
||||
queueMemorySync(TEST_PROJECT, `mem-${i}`, memories[i]);
|
||||
}
|
||||
|
||||
const result = await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
expect(result.synced).toBe(3);
|
||||
});
|
||||
|
||||
it("preserves_memory_content_through_queue", async () => {
|
||||
const memory = {
|
||||
type: "note",
|
||||
content: "Important decision",
|
||||
projectId: TEST_PROJECT,
|
||||
metadata: { createdAt: "2024-01-01" },
|
||||
};
|
||||
|
||||
queueMemorySync(TEST_PROJECT, "mem-1", memory);
|
||||
|
||||
const status = getSyncStatus(TEST_PROJECT);
|
||||
expect(status.queued).toBe(1);
|
||||
|
||||
await flushSyncQueue(TEST_PROJECT);
|
||||
|
||||
// Should complete without data loss
|
||||
expect(status.synced).toBe(0); // Old status object
|
||||
const newStatus = getSyncStatus(TEST_PROJECT);
|
||||
expect(newStatus.synced).toBe(1);
|
||||
});
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Reference in a new issue