Concurrency fixes. Docs.

.gitignore

@@ -34,4 +34,8 @@ bunfig.toml
 
 # Local
 lab/
-lab.ts
+lab.ts
+mydatabase*/
+
+# Pup data
+.pup

README.md

@@ -4,10 +4,10 @@ A cross-platform, in-memory indexed and file based Key/Value database for
 JavaScript and TypeScript, designed for seamless multi-process access and
 compatibility across Node.js, Deno, and Bun.
 
-_Please note that `cross/kv` is still under development. The API and features
-are starting to stabilize, but are still subject to change._
+_Please note that `cross/kv` is currently in **beta**. The API and features are
+starting to stabilize, but are still subject to change._
 
-## **Features**
+## Features
 
 - **Indexed Key/Value Storage**: Store and retrieve data easily using
   hierarchical keys, with an in-memory index to provide fast lookups of large
@@ -24,7 +24,7 @@ are starting to stabilize, but are still subject to change._
 - **Key Ranges:** Retrieve ranges of data efficiently directly from the index
   using key ranges.
 
-## **Installation**
+## Installation
 
 Full installation instructions available at <https://jsr.io/@cross/kv>
 
@@ -39,12 +39,13 @@ deno add @cross/kv
 bunx jsr add @cross/kv
 ```
 
-## **Simple Usage**
+## Simple Usage
 
 ```typescript
 import { KV } from "@cross/kv";
 
 const kvStore = new KV();
+
 await kvStore.open("./mydatabase/"); // Path where data files will be stored
 
 // Set a value
@@ -61,7 +62,7 @@ await kvStore.delete(["data", "username"]);
 await kvStore.close();
 ```
 
-## **Advanced Usage**
+## Advanced Usage
 
 ```typescript
 import { KV } from "@cross/kv";
@@ -118,20 +119,24 @@ console.log("Ben: ", ben); // Outputs the object of Ben
 await kvStore.close();
 ```
 
-## **API Documentation**
+## API Documentation
 
 ### Methods
 
-- `KV` class
+- `KV(options)` - Main class. Options such as `autoSync` and `syncIntervalMs`
+  are optional.
   - `async open(filepath)` - Opens the KV store.
   - `async set(key, value)` - Stores a value.
   - `async get(key)` - Retrieves a value.
   - `async *iterate(query)` - Iterates over entries for a key.
   - `async listAll(query)` - Gets all entries for a key as an array.
   - `delete(key)` - Deletes a key-value pair.
   - `beginTransaction()` - Starts a transaction.
-  - `async endTransaction()` - Ends a transaction.
+  - `async endTransaction()` - Ends a transaction, returns a list of `Errors` if
+    any occurred.
   - `async vacuum()` - Reclaims storage space.
+  - `on(eventName, eventData)` - Listen for events such as `sync`,
+    `watchdogError` or `closing`.
   - `close()` - Closes the KV store.
 
 ### Keys
@@ -185,6 +190,58 @@ objects like `{ from, to }`. An empty range (`{}`) means any document.
 ["products", "book", {}, "author"]
 ```
 
+## Multi-Process Synchronization
+
+`cross/kv` has a built in mechanism for synchronizing the in-memory index with
+the transaction ledger, allowing for multiple processes to work with the same
+database simultanously. Due to the append-only design of the ledger, each
+process can update it's internal state by reading everything after the last
+processed transaction. An internal watchdog actively checks for new transactions
+and updates the in-memory index accordingly. The synchnization frequency can be
+controlled by the option `syncIntervalMs`, which defaults to `1000` (1 second).
+
+In single process scenarios, the watchdog can be disabled by setting the
+`autoSync` option to `false`.
+
+Subscribe to the `sync` event to receive notifications about synchronization
+results and potential errors.
+
+```typescript
+const kvStore = new KV();
+await kvStore.open("./mydatabase/");
+
+// Subscribe to sync events for monitoring
+kvStore.on("sync", (eventData) => {
+  switch (eventData.result) {
+    case "ready":
+      console.log("Everything is up to date.");
+      break;
+    case "blocked":
+      console.warn(
+        "Synchronization is temporarily blocked (e.g., during vacuum).",
+      );
+      break;
+    case "success":
+      console.log(
+        "Synchronization completed successfully, new transactions added to the index.",
+      );
+      break;
+    case "ledgerInvalidated":
+      console.warn(
+        "Ledger invalidated! The database hash been reopened and the index resynchronized to maintain consistency.",
+      );
+      break;
+    case "error":
+      // Error Handling
+      console.error("Synchronization error:", eventData.error);
+      // Log the error, report it, or take appropriate action.
+      break;
+    default:
+      console.warn("Unknown sync result:", eventData.result);
+  }
+});
+```
+
 ## **Contributing**
 
 Contributions are welcome! Feel free to open issues or submit pull requests.

deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@cross/kv",
-  "version": "0.0.11",
+  "version": "0.0.12",
   "exports": {
     ".": "./mod.ts"
   },

src/constants.ts

@@ -1,11 +1,13 @@
-export const LOCK_DEFAULT_MAX_RETRIES = 32;
-export const LOCK_DEFAULT_INITIAL_RETRY_INTERVAL_MS = 20; // Increased with itself on each retry, so the actual retry interval is 20, 40, 60 etc. 32 and 20 become about 10 seconds total.
+export const LOCK_DEFAULT_MAX_RETRIES = 15;
+export const LOCK_DEFAULT_INITIAL_RETRY_INTERVAL_MS = 100; // Increased with itself on each retry, so the actual retry interval is 20, 40, 60 etc. 32 and 20 become about 10 seconds total.
 export const LOCK_STALE_TIMEOUT_S = 60_000;
 
 export const SUPPORTED_LEDGER_VERSIONS = ["ALPH"];
 
 export const LEDGER_BASE_OFFSET = 1_024;
 
-export const LEDGER_PREFETCH_BYTES = 1_024;
+export const LEDGER_MAX_READ_FAILURES = 10;
+
+export const LEDGER_PREFETCH_BYTES = 16000;
 
 export const SYNC_INTERVAL_MS = 1_000;

src/kv.test.ts

@@ -1,7 +1,8 @@
-import { assertEquals, assertRejects } from "@std/assert";
+import { assertEquals, assertRejects, assertThrows } from "@std/assert";
 import { test } from "@cross/test";
-import { KV, type KVDataEntry } from "./kv.ts";
+import { KV, type KVDataEntry, KVOptions } from "./kv.ts";
 import { tempfile } from "@cross/fs";
+import { SYNC_INTERVAL_MS } from "./constants.ts";
 
 test("KV: set, get and delete (numbers and strings)", async () => {
   const tempFilePrefix = await tempfile();
@@ -289,3 +290,67 @@ test("KV: vacuum", async () => {
 
   kvStore.close();
 });
+
+test("KV Options: defaults work correctly", () => {
+  const kv = new KV(); // No options provided
+  assertEquals(kv.autoSync, true);
+  assertEquals(kv.syncIntervalMs, SYNC_INTERVAL_MS);
+  kv.close();
+});
+
+test("KV Options: custom options are applied", () => {
+  const options: KVOptions = {
+    autoSync: false,
+    syncIntervalMs: 5000,
+  };
+  const kv = new KV(options);
+  assertEquals(kv.autoSync, false);
+  assertEquals(kv.syncIntervalMs, 5000);
+  kv.close();
+});
+
+test("KV Options: throws on invalid autoSync type", () => {
+  const options: KVOptions = {
+    // @ts-expect-error Test
+    autoSync: "not a boolean", // Incorrect type
+  };
+  assertThrows(
+    () => new KV(options),
+    TypeError,
+    "Invalid option: autoSync must be a boolean",
+  );
+});
+
+test("KV Options: throws on invalid syncIntervalMs type", () => {
+  const options: KVOptions = {
+    // @ts-expect-error Test
+    syncIntervalMs: "not a number", // Incorrect type
+  };
+  assertThrows(
+    () => new KV(options),
+    TypeError,
+    "Invalid option: syncIntervalMs must be a positive integer",
+  );
+});
+
+test("KV Options: throws on negative syncIntervalMs", () => {
+  const options: KVOptions = {
+    syncIntervalMs: -1000, // Negative value
+  };
+  assertThrows(
+    () => new KV(options),
+    TypeError,
+    "Invalid option: syncIntervalMs must be a positive integer",
+  );
+});
+
+test("KV Options: throws on zero syncIntervalMs", () => {
+  const options: KVOptions = {
+    syncIntervalMs: 0, // Zero value
+  };
+  assertThrows(
+    () => new KV(options),
+    TypeError,
+    "Invalid option: syncIntervalMs must be a positive integer",
+  );
+});