search: a fragment's own index, opt-in (the capability)

.github/workflows/ci.yml

@@ -208,7 +208,8 @@ jobs:
     needs: proven
     if: ${{ !cancelled() && (needs.proven.result == 'skipped' || (needs.proven.result == 'success' && needs.proven.outputs.proven == 'false')) }}
     runs-on: ubuntu-latest
-    timeout-minutes: 20
+    # shard 1 builds the search service's image (below) when search/ moved
+    timeout-minutes: ${{ matrix.shard == 1 && 50 || 20 }}
     env:
       # xtask/src/build.rs, WORKER_BUILD_VERSION
       WORKER_BUILD_VERSION: 0.8.5
@@ -225,6 +226,28 @@ jobs:
         with:
           save: ${{ matrix.shard == 1 && github.event_name == 'push' && github.ref == 'refs/heads/master' }}
           worker-build-version: ${{ env.WORKER_BUILD_VERSION }}
+      # The search service's image (search/Dockerfile) for the `search`
+      # section, on shard 1 alone, which runs it: loaded into Docker as the
+      # section names it, its layers BuildKit's in the Actions cache (scope
+      # e2e-search). An unchanged search/ is its layers fetched; a changed
+      # one compiles LanceDB here (about 25 minutes on a runner), hence
+      # shard 1's longer limit.
+      - if: matrix.shard == 1
+        uses: docker/build-push-action@c3c9e263c25d99ce0380d002d59b67737d91b0dc # v7.4.0
+        env:
+          DOCKER_BUILD_SUMMARY: false
+          DOCKER_BUILD_RECORD_UPLOAD: false
+        with:
+          builder: ${{ steps.build.outputs.builder }}
+          context: search
+          file: search/Dockerfile
+          platforms: linux/amd64
+          provenance: false
+          load: true
+          tags: fragment-search:e2e
+          build-args: JOBS=4
+          cache-from: type=gha,scope=e2e-search
+          cache-to: type=gha,scope=e2e-search,mode=max,ignore-error=true
       # one shard's sections on a fresh node; the run's scratch (each node
       # boot's log, the lanes' screenshots) is kept when a check fails, and
       # uploaded below

.github/workflows/search.yml

@@ -0,0 +1,40 @@
+# The search service's Rust (search/, search/README.md): its own workspace,
+# so its own job: host tests and clippy, warnings denied, on every change
+# here. Its object-store test (`on_an_object_store`) needs a store, and
+# runs by hand. Its Rust cache is Swatinem/rust-cache's, saved only from
+# master's pushes, as images.yml's.
+name: search
+
+on:
+  push:
+    branches: [master]
+    paths: ["search/**", ".github/workflows/search.yml"]
+  pull_request:
+    paths: ["search/**", ".github/workflows/search.yml"]
+  workflow_dispatch:
+
+concurrency:
+  group: search-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  check:
+    name: search check
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: search
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: clippy
+      - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends protobuf-compiler libprotobuf-dev
+      - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2
+        with:
+          shared-key: search
+          workspaces: search -> target
+          cache-bin: false
+          save-if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/master' }}
+      - run: cargo clippy --all-targets --locked -- -D warnings
+      - run: cargo test --locked

AGENTS.md

@@ -37,6 +37,9 @@ platform concept.
   `crates/fakes`, `crates/devstack`, `crates/e2e`: local vendor fakes and harness.
 - `templates/`: fragment templates; `applib/`: reusable app code.
 - `images/`: a separate Rust workspace (bridge, stub, Hermes).
+- `search/`: the search service (LanceDB on R2, S3 or a disk;
+  `search/README.md`); `bench/search/`: the benchmark that chose it. Each a
+  separate Rust workspace (`cargo test --release`).
 - `.github/workflows/`: CI, image promotion and CLI releases. Read the
   relevant workflow and xtask module for mechanics and flags.
   Released CLIs warn on stderr when finite.chat advertises a newer

cell/src/config.rs

@@ -100,6 +100,10 @@ pub struct Config {
     /// AI binding's input to `<url>/run/<model>` instead (a fake at the
     /// vendor boundary, labeled so: models.rs) and needs no gateway.
     pub ai_url: Option<String>,
+    /// `FRAGMENT_SEARCH_URL`: the search service (search/README.md;
+    /// searchable.rs), beside the deployment, called with `SEARCH_TOKEN`;
+    /// none: this deployment has no search, and its routes answer 404.
+    pub search_url: Option<String>,
     /// `FRAGMENT_MAIL_FROM`: a bare address or unquoted `Name <address>`
     /// (the deploy config's `mail_from`), on a domain onboarded to Email
     /// Sending; none: the deployment sends no mail (mail.rs).
@@ -354,6 +358,7 @@ impl Config {
                 assert!(id != "default", "AI_GATEWAY_ID names the deployment's own gateway: `default` makes one that logs (spike S4)");
             }),
             ai_url: var(env, "FRAGMENT_AI_URL").map(|u| u.trim_end_matches('/').to_string()),
+            search_url: var(env, "FRAGMENT_SEARCH_URL").map(|u| u.trim_end_matches('/').to_string()),
             mail_from: env.var("FRAGMENT_MAIL_FROM").ok().map(|f| {
                 fragment_core::mail::Sender::parse(&f.to_string()).unwrap_or_else(|e| panic!("FRAGMENT_MAIL_FROM: {e}"))
             }),

cell/src/ended.rs

@@ -358,6 +358,8 @@ impl FragmentCell {
         for s in due {
             let cleared = async {
                 js::delete_app_facet(&self.raw, &s.facet).await?;
+                // its search index, with its database (searchable.rs)
+                self.search_drop(s.incarnation).await?;
                 self.delete_blobs_under(&format!("{}/", s.npub), BLOB_PAGES_PER_PASS).await
             }
             .await;

cell/src/fragment.rs

@@ -97,6 +97,9 @@ pub struct FragmentCell {
     pub(crate) swept: Cell<bool>,
     /// The ledger ids a call or a sweep is settling now (channels.rs).
     pub(crate) settling: RefCell<BTreeMap<String, Arc<futures_util::lock::Mutex<()>>>>,
+    /// The search sources a write or a forget is sending now
+    /// (searchable.rs): one at a time a source.
+    pub(crate) search_sending: RefCell<BTreeMap<String, Arc<futures_util::lock::Mutex<()>>>>,
     /// How the app facet starts (ops.rs `app_loader`).
     pub(crate) app: js::AppLoader,
     /// What this activation knows of its live sockets (live.rs).
@@ -141,6 +144,7 @@ impl DurableObject for FragmentCell {
             rate: RefCell::new(rate),
             swept: Cell::new(false),
             settling: RefCell::default(),
+            search_sending: RefCell::default(),
             app,
             live: RefCell::default(),
             meter_armed: Cell::new(-1),
@@ -341,6 +345,9 @@ pub(crate) enum MetaKey {
     /// The channels its people's search holds, as the rows last named
     /// them (search.rs `search_fence`): a JSON array of names.
     Searched,
+    /// Whether the search service holds this life's sources (searchable.rs):
+    /// written at its first upsert; a push's deletes are then forgotten.
+    Search,
     /// Why live's code was not installed.
     CodeError,
     /// When the blob collection runs next.
@@ -432,6 +439,7 @@ impl MetaKey {
             MetaKey::ListedSentAt => "listed_sent_at",
             MetaKey::ListedDue => "listed_due",
             MetaKey::Searched => "searched",
+            MetaKey::Search => "search",
             MetaKey::CodeError => "code_error",
             MetaKey::BlobsGcAt => "blobs_gc_at",
             MetaKey::Vapid => "vapid",
@@ -1079,6 +1087,20 @@ impl FragmentCell {
             (Method::Delete, ["api", "schedules", id]) => self.remove_schedule(&caller, id),
             // its app's database, for its owner (export.rs)
             (Method::Get, ["api", "database"]) => self.database(&caller, query("table"), query("offset")).await,
+            (Method::Post, ["api", "search"]) => {
+                let q = body_json(&mut req).await?;
+                self.search_query(&caller, q).await
+            }
+            (Method::Post, ["api", "search", "upsert"]) => {
+                let up = body_json(&mut req).await?;
+                self.search_upsert(&caller, up).await
+            }
+            (Method::Post, ["api", "search", "delete"]) => {
+                let d = body_json(&mut req).await?;
+                self.search_delete(&caller, d).await
+            }
+            (Method::Get, ["api", "search", "sources"]) => self.search_sources(&caller, query("after"), query("limit")).await,
+            (Method::Get, ["api", "search", "status"]) => self.search_status(&caller).await,
             (Method::Post, ["api", "subscriptions"]) => {
                 let body: Value = body_json(&mut req).await?;
                 self.subscribe(&caller, &body)
@@ -1318,6 +1340,7 @@ impl FragmentCell {
         }
         self.flush_index().await;
         self.flush_search().await;
+        self.search_forget_due().await;
         self.flush_joined().await;
         if let Err(e) = self.seed().await {
             self.event("template.failed", &e.message, json!({ "code": e.code }));
@@ -1446,7 +1469,7 @@ impl FragmentCell {
             self.set_meta(MetaKey::PollAt, &poll_at.to_string())?;
         }
         let outbox = self.rows("SELECT MIN(next_at) AS at FROM index_outbox", vec![])?.first().and_then(|r| r["at"].as_i64());
-        let due = [outbox, self.listed_due_at()?, self.search_due_at()?, self.joined_due_at()?, self.runs_due_at()?, self.pending_due_at()?, self.outbox_due_at()?, self.meter_due_at()?, self.card_due_at()?, self.invites_due_at()?, ended, also];
+        let due = [outbox, self.listed_due_at()?, self.search_due_at()?, self.search_forget_due_at()?, self.joined_due_at()?, self.runs_due_at()?, self.pending_due_at()?, self.outbox_due_at()?, self.meter_due_at()?, self.card_due_at()?, self.invites_due_at()?, ended, also];
 
         let at = due.into_iter().flatten().fold(poll_at, i64::min).max(js::now_ms() + min_ms);
         self.state.storage().set_alarm(ScheduledTime::new(js_sys::Date::new(&JsValue::from_f64(at as f64)))).await?;

cell/src/keys.rs

@@ -191,6 +191,12 @@ impl WorkOs<'_> {
     }
 }
 
+/// The search service's bearer token (bound as `SEARCH_TOKEN`), sent only
+/// to the service this deployment names (`FRAGMENT_SEARCH_URL`).
+pub async fn search_token(env: &Env) -> CellResult<String> {
+    required(env, store::SEARCH_TOKEN).await
+}
+
 /// The account's restricted Stripe key (bound as `STRIPE_KEY`).
 pub async fn stripe_key(env: &Env) -> CellResult<String> {
     required(env, store::STRIPE_KEY).await

cell/src/lib.rs

@@ -69,6 +69,7 @@ mod routed;
 mod schedules;
 mod schema;
 mod search;
+mod searchable;
 mod runs_on;
 mod serve;
 mod share;
@@ -1382,7 +1383,9 @@ async fn platform_answer(mut req: Request, env: &Env, ctx: &Context, cfg: &'stat
                 (_, [] | [""]) => return Err(CellError::new(ErrorCode::NotFound, format!("no route {path}"))),
                 _ => format!("/api/{}", rest.join("/")),
             };
-            let body = read_body(&mut req, limits::BODY_MAX_BYTES).await?;
+            // a search source is written whole, so its upsert's body is larger
+            let max = if rest == ["search", "upsert"] { fragment_proto::search::UPSERT_MAX_BYTES } else { limits::BODY_MAX_BYTES };
+            let body = read_body(&mut req, max).await?;
             // the inbox takes its token instead of a signature
             let mut extra = vec![];
             let principal = match inner.as_str() {

cell/src/plane.rs

@@ -48,6 +48,8 @@ pub struct PinMove {
     pub changed: bool,
     pub to: Option<String>,
     pub paths: Vec<String>,
+    /// The paths it deleted (of `paths`).
+    pub removed: Vec<String>,
 }
 
 /// A file in a pin's tree index (metadata only; the bytes stay in git).
@@ -258,16 +260,16 @@ impl FragmentCell {
         let cs = self.cs()?;
         let Some(to) = cs.branch_head(&repo, which).await? else {
             if from.is_none() {
-                return Ok(PinMove { changed: false, to: None, paths: vec![] });
+                return Ok(PinMove { changed: false, to: None, paths: vec![], removed: vec![] });
             }
             self.exec("DELETE FROM tree WHERE ref = ?", vec![which.into()])?;
             self.exec("DELETE FROM pointers WHERE ref = ?", vec![which.into()])?;
             self.del_meta(MetaKey::pin(which))?;
             self.event("git.refresh", &format!("{which}: the branch is gone; pin cleared"), json!({ "ref": which, "from": from }));
-            return Ok(PinMove { changed: true, to: None, paths: vec![] });
+            return Ok(PinMove { changed: true, to: None, paths: vec![], removed: vec![] });
         };
         if from.as_deref() == Some(to.as_str()) {
-            return Ok(PinMove { changed: false, to: Some(to), paths: vec![] });
+            return Ok(PinMove { changed: false, to: Some(to), paths: vec![], removed: vec![] });
         }
         let entries = cs.tree(&repo, &to).await?;
         let prior: BTreeMap<String, Indexed> =
@@ -283,7 +285,8 @@ impl FragmentCell {
             &format!("{which}: {} → {} ({} files, {} changed)", short(from.as_deref()), short(Some(&to)), entries.len(), paths.len()),
             json!({ "ref": which, "from": from, "to": to, "changed": paths.iter().take(50).collect::<Vec<_>>() }),
         );
-        Ok(PinMove { changed: true, to: Some(to), paths })
+        let removed = diff.removed.iter().map(|p| p.to_string()).collect();
+        Ok(PinMove { changed: true, to: Some(to), paths, removed })
     }
 
     /// Writes a pin move's changes to its index (`tree::diff`): deletes
@@ -662,6 +665,7 @@ impl FragmentCell {
                 self.broadcast_change(moved.to.as_deref(), &moved.paths);
                 let depth = self.commit_depth(moved.to.as_deref())?;
                 self.fire_files(moved.to.as_deref(), &moved.paths, depth, read_only)?;
+                self.search_forget(&moved.removed)?;
             }
             self.follow(which).await?;
             out.push((which.to_string(), moved));