//! Provider-neutral index jobs and vector-space registry.
use crate::errors::{MCSError, Result};
use crate::events::{Lease, lease_until, parse_uuid, sha256, sql_error};
use crate::graph::TxGuard;
use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
/// Profile ids that must receive an index-job update. When no managed
/// profile serves the store, the list holds the nil id so the job is held.
pub(crate) fn serving_profile_ids(conn: &Connection) -> Result<Vec<Uuid>> {
let mut stmt = conn.prepare("SELECT serving_profile FROM index_profile_registry WHERE serving_profile IS NOT NULL UNION SELECT candidate_profile FROM index_profile_registry WHERE state='Rebuilding' AND candidate_profile IS NOT NULL").map_err(sql_error)?;
let mut profiles = stmt
.query_map([], |r| r.get::<_, String>(0))
.map_err(sql_error)?
.collect::<rusqlite::Result<Vec<_>>>()
.map_err(sql_error)?;
if profiles.is_empty() {
profiles.push(uuid::Uuid::nil().to_string());
}
profiles.iter().map(|profile| parse_uuid(profile)).collect()
}
/// Queue one owner for every serving profile, keyed on `chunk_index_job`
/// `(profile_id, owner_kind, owner_id)`. A nil profile is an explicitly held
/// job, never a claimable provider profile; managed serving and candidate
/// profiles receive pending rows. One row per owner and profile.
pub(crate) fn enqueue_chunk_change(
conn: &Connection,
owner_kind: OwnerKind,
owner_id: i64,
revision: i64,
deleted: bool,
) -> Result<()> {
let profiles = serving_profile_ids(conn)?;
for profile in profiles {
let state = if profile.is_nil() { "held" } else { "pending" };
conn.execute(
"INSERT INTO chunk_index_job(profile_id,owner_kind,owner_id,owner_revision,operation,state)
VALUES(?1,?2,?3,?4,?5,?6)
ON CONFLICT(profile_id,owner_kind,owner_id) DO UPDATE SET
owner_revision=excluded.owner_revision,operation=excluded.operation,state=excluded.state,
lease_token=NULL,lease_epoch=lease_epoch+1,lease_until_us=0,attempts=0,
next_attempt_us=0,last_error=NULL",
params![
profile.to_string(),
owner_kind.as_str(),
owner_id,
revision,
if deleted { "delete" } else { "upsert" },
state
],
)
.map_err(sql_error)?;
conn.execute(
"UPDATE ann_generation SET full_scan_generation=NULL WHERE profile_id=?1",
[profile.to_string()],
)
.map_err(sql_error)?;
}
Ok(())
}
/// The entity path of [`enqueue_chunk_change`], kept as the narrow wrapper so
/// the change-event hook in `events.rs` keeps its entity-only signature. The
/// legacy `index_job` table was dropped by migration 0009.
pub(crate) fn enqueue_change(
conn: &Connection,
entity_id: i64,
revision: i64,
deleted: bool,
) -> Result<()> {
enqueue_chunk_change(conn, OwnerKind::Entity, entity_id, revision, deleted)
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum Normalization {
None,
L2,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum ChunkKind {
Identity,
Observation,
Relation,
}
impl ChunkKind {
pub const fn as_str(self) -> &'static str {
match self {
ChunkKind::Identity => "identity",
ChunkKind::Observation => "observation",
ChunkKind::Relation => "relation",
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum OwnerKind {
Entity,
Relation,
}
impl OwnerKind {
pub const fn as_str(self) -> &'static str {
match self {
OwnerKind::Entity => "entity",
OwnerKind::Relation => "relation",
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum DistanceMetric {
Cosine,
InnerProduct,
L2Squared,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct IndexProfile {
pub id: Uuid,
pub store_key: String,
pub provider_kind: String,
pub model: String,
pub dimensions: u32,
pub representation_version: String,
pub normalization: Normalization,
pub distance_metric: DistanceMetric,
pub vector_encoding_version: String,
}
impl IndexProfile {
pub fn validate(&self) -> Result<()> {
if self.id.is_nil()
|| self.store_key != "default"
|| self.dimensions == 0
|| self.dimensions > 65_536
|| self.vector_encoding_version != "f32le-v1"
|| [
&self.provider_kind,
&self.model,
&self.representation_version,
]
.iter()
.any(|s| s.trim().is_empty() || s.len() > 256 || s.chars().any(char::is_control))
{
return Err(MCSError::InvalidParams("invalid index profile".into()));
}
Ok(())
}
/// serde_json's default sorted object map is the canonical key order.
pub fn fingerprint(&self) -> Result<String> {
self.validate()?;
let mut value = serde_json::to_value(self)?;
value
.as_object_mut()
.ok_or_else(|| MCSError::MemoryError("profile must be an object".into()))?
.remove("id");
Ok(sha256(&serde_json::to_vec(&value)?))
}
pub fn validate_vector(&self, vector: &[f32]) -> Result<()> {
if vector.len() != self.dimensions as usize || vector.iter().any(|x| !x.is_finite()) {
return Err(MCSError::InvalidParams(
"vector dimensions or finite-value validation failed".into(),
));
}
if self.normalization == Normalization::L2 {
let norm: f64 = vector.iter().map(|x| f64::from(*x).powi(2)).sum();
if (norm - 1.0).abs() > 1e-4 {
return Err(MCSError::InvalidParams(
"vector is not L2 normalized".into(),
));
}
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum StoreState {
LegacyCompat,
Active(Uuid),
Rebuilding {
serving: Option<Uuid>,
candidate: Uuid,
},
Failed {
serving: Option<Uuid>,
candidate: Uuid,
reason: String,
},
}
pub struct IndexProfileRegistry<'a> {
conn: &'a Connection,
}
impl<'a> IndexProfileRegistry<'a> {
pub const fn new(conn: &'a Connection) -> Self {
Self { conn }
}
/// The connection this registry reads. The server crate serves taxonomy
/// snapshots through it: the generation read, the vector rows and the
/// publish mark then share one transaction view with the registry state.
pub const fn connection(&self) -> &Connection {
self.conn
}
pub fn get(&self, id: Uuid) -> Result<IndexProfile> {
let text: String = self
.conn
.query_row(
"SELECT definition FROM index_profile WHERE id=?1",
[id.to_string()],
|r| r.get(0),
)
.map_err(sql_error)?;
Ok(serde_json::from_str(&text)?)
}
pub fn state(&self, store_key: &str) -> Result<StoreState> {
let (state,serving,candidate,reason): (String,Option<String>,Option<String>,Option<String>) = self.conn.query_row("SELECT state,serving_profile,candidate_profile,failure_reason FROM index_profile_registry WHERE store_key=?1", [store_key], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?))).map_err(sql_error)?;
let serving = serving.as_deref().map(parse_uuid).transpose()?;
let candidate = candidate.as_deref().map(parse_uuid).transpose()?;
match (state.as_str(), serving, candidate, reason) {
("LegacyCompat", None, None, None) => Ok(StoreState::LegacyCompat),
("Active", Some(profile), None, None) => Ok(StoreState::Active(profile)),
("Rebuilding", serving, Some(candidate), None) => {
Ok(StoreState::Rebuilding { serving, candidate })
}
("Failed", serving, Some(candidate), Some(reason)) => Ok(StoreState::Failed {
serving,
candidate,
reason,
}),
_ => Err(MCSError::MemoryError(
"invalid persisted profile registry state".into(),
)),
}
}
pub fn serving_profile(&self, store_key: &str) -> Result<Option<Uuid>> {
Ok(match self.state(store_key)? {
StoreState::LegacyCompat => None,
StoreState::Active(profile) => Some(profile),
StoreState::Rebuilding { serving, .. } | StoreState::Failed { serving, .. } => serving,
})
}
pub fn begin_rebuild(&self, profile: &IndexProfile) -> Result<()> {
let fingerprint = profile.fingerprint()?;
let tx = TxGuard::begin(self.conn)?;
if matches!(
self.state(&profile.store_key)?,
StoreState::Rebuilding { .. }
) {
return Err(MCSError::InvalidParams(
"profile rebuild already in progress".into(),
));
}
if self.serving_profile(&profile.store_key)? == Some(profile.id) {
return Err(MCSError::InvalidParams(
"cannot rebuild into serving profile".into(),
));
}
// Profiles are immutable. Reusing a retired/candidate ID would also
// reuse its generation and vectors, defeating the rebuild boundary.
self.conn
.execute(
"INSERT INTO index_profile VALUES(?1,?2,?3,?4,'Rebuilding')",
params![
profile.id.to_string(),
profile.store_key,
fingerprint,
serde_json::to_string(profile)?
],
)
.map_err(sql_error)?;
self.conn.execute("UPDATE index_profile SET state='Retired' WHERE id=(SELECT candidate_profile FROM index_profile_registry WHERE store_key=?1)", [&profile.store_key]).map_err(sql_error)?;
self.conn.execute("UPDATE index_profile_registry SET state='Rebuilding',candidate_profile=?2,failure_reason=NULL WHERE store_key=?1", params![profile.store_key,profile.id.to_string()]).map_err(sql_error)?;
self.conn
.execute(
"INSERT INTO ann_generation(profile_id) VALUES(?1)",
[profile.id.to_string()],
)
.map_err(sql_error)?;
self.conn.execute("INSERT INTO entity_revision SELECT id,1,0 FROM entity WHERE flags=0 ON CONFLICT(entity_id) DO NOTHING", []).map_err(sql_error)?;
self.conn.execute("INSERT INTO chunk_index_job(profile_id,owner_kind,owner_id,owner_revision,operation) SELECT ?1,'entity',e.id,r.revision,'upsert' FROM entity e JOIN entity_revision r ON r.entity_id=e.id WHERE e.flags=0", [profile.id.to_string()]).map_err(sql_error)?;
// Every relation mirror joins the rebuild: live relations embed as
// upserts and tombstoned mirrors re-run as deletes, so a rebuild
// also purges orphan chunk rows for relations that no longer exist.
self.conn.execute("INSERT INTO chunk_index_job(profile_id,owner_kind,owner_id,owner_revision,operation) SELECT ?1,'relation',m.id,m.revision,CASE WHEN m.deleted=0 THEN 'upsert' ELSE 'delete' END FROM taxonomy_relation m", [profile.id.to_string()]).map_err(sql_error)?;
tx.commit()
}
pub fn fail_rebuild(&self, candidate: Uuid, reason: &str) -> Result<()> {
let tx = TxGuard::begin(self.conn)?;
let changed = self.conn.execute("UPDATE index_profile_registry SET state='Failed',failure_reason=?2 WHERE state='Rebuilding' AND candidate_profile=?1", params![candidate.to_string(),reason.chars().take(2048).collect::<String>()]).map_err(sql_error)?;
if changed != 1 {
return Err(MCSError::InvalidParams(
"candidate is not rebuilding".into(),
));
}
self.conn.execute("UPDATE chunk_index_job SET state='held',lease_token=NULL,lease_epoch=lease_epoch+1 WHERE profile_id=?1 AND state!='done'", [candidate.to_string()]).map_err(sql_error)?;
tx.commit()
}
pub fn activate(&self, candidate: Uuid) -> Result<()> {
let tx = TxGuard::begin(self.conn)?;
let profile = self.get(candidate)?;
if !matches!(self.state(&profile.store_key)?, StoreState::Rebuilding {candidate: c,..} if c==candidate)
{
return Err(MCSError::InvalidParams(
"candidate is not rebuilding".into(),
));
}
verify_vectors_current(self.conn, candidate)?;
let generation = AnnGenerationRepository::new(self.conn).get(candidate)?;
if generation.full_scan_generation != Some(generation.durable_generation)
|| generation.published_generation != generation.durable_generation
{
return Err(MCSError::InvalidParams(
"candidate requires verified Full scan and matching published ANN generation"
.into(),
));
}
self.conn
.execute(
"UPDATE index_profile SET state='Retired' WHERE store_key=?1 AND state='Active'",
[&profile.store_key],
)
.map_err(sql_error)?;
self.conn
.execute(
"UPDATE index_profile SET state='Active' WHERE id=?1",
[candidate.to_string()],
)
.map_err(sql_error)?;
self.conn.execute("UPDATE index_profile_registry SET state='Active',serving_profile=?2,candidate_profile=NULL,failure_reason=NULL WHERE store_key=?1", params![profile.store_key,candidate.to_string()]).map_err(sql_error)?;
self.conn.execute("UPDATE chunk_index_job SET state='held',lease_token=NULL,lease_epoch=lease_epoch+1 WHERE profile_id!=?1 AND state!='done'", [candidate.to_string()]).map_err(sql_error)?;
tx.commit()
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum IndexOperation {
Upsert,
Delete,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct IndexJob {
pub profile_id: Uuid,
pub operation: IndexOperation,
pub owner_kind: OwnerKind,
pub owner_id: i64,
pub owner_revision: i64,
pub lease: Lease,
pub attempts: i64,
}
pub struct IndexJobRepository<'a> {
conn: &'a Connection,
}
impl<'a> IndexJobRepository<'a> {
pub const fn new(conn: &'a Connection) -> Self {
Self { conn }
}
pub fn claim_due(&self, now: i64, duration_us: i64) -> Result<Option<IndexJob>> {
let until = lease_until(now, duration_us)?;
let tx = TxGuard::begin(self.conn)?;
let row: Option<(String,i64,String,i64,String,i64,i64)> = self.conn.query_row("SELECT owner_kind,owner_id,profile_id,owner_revision,operation,lease_epoch,attempts FROM chunk_index_job j WHERE ((state='pending' AND next_attempt_us<=?1) OR (state='leased' AND lease_until_us<=?1)) AND EXISTS(SELECT 1 FROM index_profile_registry r WHERE r.serving_profile=j.profile_id OR (r.state='Rebuilding' AND r.candidate_profile=j.profile_id)) ORDER BY next_attempt_us,owner_kind,owner_id,profile_id LIMIT 1", [now], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?,r.get(6)?))).optional().map_err(sql_error)?;
let job = row.map(|(owner_kind,owner_id,profile,revision,operation,epoch,attempts)| -> Result<IndexJob> {
let token = Uuid::new_v4();
self.conn.execute("UPDATE chunk_index_job SET state='leased',lease_token=?4,lease_epoch=lease_epoch+1,lease_until_us=?5,attempts=attempts+1 WHERE profile_id=?1 AND owner_kind=?2 AND owner_id=?3", params![profile,owner_kind,owner_id,token.to_string(),until]).map_err(sql_error)?;
Ok(IndexJob { profile_id:parse_uuid(&profile)?,operation:match operation.as_str() { "upsert"=>IndexOperation::Upsert,"delete"=>IndexOperation::Delete,_=>return Err(MCSError::MemoryError("invalid index operation".into())) },owner_kind:match owner_kind.as_str() { "entity"=>OwnerKind::Entity,"relation"=>OwnerKind::Relation,_=>return Err(MCSError::MemoryError("invalid owner kind".into())) },owner_id,owner_revision:revision,lease:Lease {token,epoch:epoch+1,until_us:until},attempts:attempts+1 })
}).transpose()?;
tx.commit()?;
Ok(job)
}
pub fn renew(&self, job: &IndexJob, now: i64, duration_us: i64) -> Result<bool> {
let until = lease_until(now, duration_us)?;
let tx = TxGuard::begin(self.conn)?;
let changed = self.conn.execute("UPDATE chunk_index_job SET lease_until_us=?7 WHERE profile_id=?1 AND owner_kind=?2 AND owner_id=?3 AND lease_token=?4 AND lease_epoch=?5 AND state='leased' AND lease_until_us>?6", params![job.profile_id.to_string(),job.owner_kind.as_str(),job.owner_id,job.lease.token.to_string(),job.lease.epoch,now,until]).map_err(sql_error)?;
tx.commit()?;
Ok(changed == 1)
}
pub fn retry(
&self,
job: &IndexJob,
now: i64,
next_attempt_us: i64,
error: &str,
dead: bool,
) -> Result<bool> {
let tx = TxGuard::begin(self.conn)?;
let changed = self.conn.execute("UPDATE chunk_index_job SET state=?7,next_attempt_us=?8,last_error=?9 WHERE profile_id=?1 AND owner_kind=?2 AND owner_id=?3 AND lease_token=?4 AND lease_epoch=?5 AND state='leased' AND lease_until_us>?6", params![job.profile_id.to_string(),job.owner_kind.as_str(),job.owner_id,job.lease.token.to_string(),job.lease.epoch,now,if dead {"dead"} else {"pending"},next_attempt_us,error.chars().take(2048).collect::<String>()]).map_err(sql_error)?;
// A dead-lettered owner must not keep a stale chunk set in the
// candidate: the verified full-scan gate would otherwise publish a
// snapshot serving an outdated embedding. Its next write re-enqueues
// the owner from scratch.
if changed == 1 && dead {
self.conn
.execute(
"DELETE FROM chunk_vector WHERE profile_id=?1 AND owner_kind=?2 AND owner_id=?3",
params![job.profile_id.to_string(), job.owner_kind.as_str(), job.owner_id],
)
.map_err(sql_error)?;
}
tx.commit()?;
Ok(changed == 1)
}
/// Fenced chunk commit: delete the owner's old chunk rows, insert the new
/// ones, and advance the durable generation in one transaction. A delete
/// operation passes `None` as the chunks and removes the rows. Every
/// vector is validated against the profile before it is stored, so a
/// wrong dimension, a NaN, or a non-unit L2 vector fails the job instead
/// of poisoning the snapshot.
pub fn commit_chunks(
&self,
job: &IndexJob,
now_us: i64,
chunks: Option<&[&(ChunkKind, &[f32])]>,
source: &str,
) -> Result<bool> {
let tx = TxGuard::begin(self.conn)?;
let current = self
.conn
.query_row(
"SELECT owner_revision, state, lease_token, lease_epoch, lease_until_us
FROM chunk_index_job WHERE profile_id=?1 AND owner_kind=?2 AND owner_id=?3",
params![
job.profile_id.to_string(),
job.owner_kind.as_str(),
job.owner_id
],
|r| {
Ok((
r.get::<_, i64>(0)?,
r.get::<_, String>(1)?,
r.get::<_, Option<String>>(2)?,
r.get::<_, i64>(3)?,
r.get::<_, i64>(4)?,
))
},
)
.optional()
.map_err(sql_error)?;
let Some((revision, state, token, epoch, until)) = current else {
return Ok(false);
};
if revision != job.owner_revision
|| state != "leased"
|| token != Some(job.lease.token.to_string())
|| epoch != job.lease.epoch
|| until <= now_us
{
return Ok(false);
}
// The fence compares the job revision against the LIVE owner revision
// (entity_revision or taxonomy_relation), so a stale job never lands.
let Some((live_revision, live_deleted)) =
self.owner_revision(self.conn, job.owner_kind, job.owner_id)?
else {
return Ok(false);
};
if live_revision != job.owner_revision {
return Ok(false);
}
match job.operation {
IndexOperation::Delete => {
if !live_deleted {
return Ok(false);
}
}
IndexOperation::Upsert => {
if live_deleted {
return Ok(false);
}
}
}
// The payload must match the operation, before any row is touched:
// an upsert without chunks would otherwise wipe the owner's rows and
// a delete with chunks would write for a tombstoned owner.
match (job.operation, chunks.is_some()) {
(IndexOperation::Upsert, true) | (IndexOperation::Delete, false) => {}
_ => {
return Err(MCSError::InvalidParams(
"chunk payload does not match job operation".into(),
));
}
}
self.conn
.execute(
"DELETE FROM chunk_vector WHERE profile_id=?1 AND owner_kind=?2 AND owner_id=?3",
params![
job.profile_id.to_string(),
job.owner_kind.as_str(),
job.owner_id
],
)
.map_err(sql_error)?;
if let Some(chunk_list) = chunks {
let profile = IndexProfileRegistry::new(self.conn).get(job.profile_id)?;
let type_id = owner_type_id(self.conn, job.owner_kind, job.owner_id)?;
for (idx, (chunk_kind, vector)) in chunk_list.iter().enumerate() {
profile.validate_vector(vector)?;
self.conn
.execute(
"INSERT INTO chunk_vector(profile_id,kind,owner_kind,owner_id,chunk_index,type_id,owner_revision,blob,created_at_us,source)
VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)",
params![
job.profile_id.to_string(),
chunk_kind.as_str(),
job.owner_kind.as_str(),
job.owner_id,
idx as i64,
type_id,
job.owner_revision,
vector.iter().flat_map(|x| x.to_le_bytes()).collect::<Vec<u8>>(),
now_us,
source,
],
)
.map_err(sql_error)?;
}
}
self.conn
.execute(
"UPDATE chunk_index_job SET state='done' WHERE profile_id=?1 AND owner_kind=?2 AND owner_id=?3",
params![
job.profile_id.to_string(),
job.owner_kind.as_str(),
job.owner_id
],
)
.map_err(sql_error)?;
self.conn
.execute(
"UPDATE ann_generation SET durable_generation=durable_generation+1,full_scan_generation=NULL WHERE profile_id=?1",
[job.profile_id.to_string()],
)
.map_err(sql_error)?;
if job.owner_kind == OwnerKind::Relation {
// Relation commits also advance the taxonomy kind-2 generation so
// the derived snapshot refreshes (Task 5 reads it). The first
// relation commit for a profile creates the kind's generation
// marker, mirroring what enqueue_taxonomy did for kinds 0/1; the
// old kind-2 funnel that created the row is retired.
self.conn
.execute(
"INSERT INTO taxonomy_ann_generation(profile_id,subject_kind) VALUES(?1,2)
ON CONFLICT(profile_id,subject_kind) DO NOTHING",
[job.profile_id.to_string()],
)
.map_err(sql_error)?;
self.conn
.execute(
"UPDATE taxonomy_ann_generation SET durable_generation=durable_generation+1,full_scan_generation=NULL
WHERE profile_id=?1 AND subject_kind=2",
[job.profile_id.to_string()],
)
.map_err(sql_error)?;
}
tx.commit()?;
Ok(true)
}
pub fn owner_revision(
&self,
conn: &Connection,
owner_kind: OwnerKind,
owner_id: i64,
) -> Result<Option<(i64, bool)>> {
match owner_kind {
OwnerKind::Entity => conn
.query_row(
"SELECT revision, deleted FROM entity_revision WHERE entity_id=?1",
[owner_id],
|r| Ok((r.get::<_, i64>(0)?, r.get::<_, bool>(1)?)),
)
.optional()
.map_err(sql_error),
OwnerKind::Relation => conn
.query_row(
"SELECT revision, deleted FROM taxonomy_relation WHERE id=?1",
[owner_id],
|r| Ok((r.get::<_, i64>(0)?, r.get::<_, bool>(1)?)),
)
.optional()
.map_err(sql_error),
}
}
}
/// The `type_id` of one owner, used to tag its chunk rows. Single-row
/// lookups mirror the source tables the owner revision fence reads.
fn owner_type_id(conn: &Connection, owner_kind: OwnerKind, owner_id: i64) -> Result<i64> {
match owner_kind {
OwnerKind::Entity => conn
.query_row("SELECT type_id FROM entity WHERE id=?1", [owner_id], |r| {
r.get(0)
})
.map_err(sql_error),
OwnerKind::Relation => conn
.query_row(
"SELECT type_id FROM taxonomy_relation WHERE id=?1",
[owner_id],
|r| r.get(0),
)
.map_err(sql_error),
}
}
fn verify_vectors_current(conn: &Connection, profile: Uuid) -> Result<()> {
// The full-scan gate accepts only a fully current chunk set across both
// owner kinds. A dead-lettered job declares its owner unindexable: the
// worker deletes the owner's chunk rows when it dead-letters, so no
// stale chunk sneaks into the snapshot, and the gate must not block the
// whole store on it. Any unfinished job also fails the scan: a pending
// owner is exactly a chunk the worker has not written yet. Rebuild
// enqueues every relation mirror, so a relation without its chunk can
// only mean the worker has not caught up.
let invalid: bool = conn
.query_row(
"SELECT EXISTS(
SELECT 1 FROM entity e
JOIN entity_revision r ON r.entity_id = e.id
LEFT JOIN chunk_vector v ON v.profile_id=?1 AND v.owner_kind='entity'
AND v.owner_id=e.id AND v.kind='identity'
WHERE e.flags=0
AND NOT EXISTS(SELECT 1 FROM chunk_index_job d
WHERE d.profile_id=?1 AND d.owner_kind='entity' AND d.owner_id=e.id
AND d.state='dead')
AND (v.owner_id IS NULL OR v.owner_revision != r.revision)
)
OR EXISTS(
SELECT 1 FROM chunk_vector v
LEFT JOIN entity e ON e.id=v.owner_id
WHERE v.profile_id=?1 AND v.owner_kind='entity'
AND (e.id IS NULL OR e.flags!=0)
)
OR EXISTS(
SELECT 1 FROM taxonomy_relation m
LEFT JOIN chunk_vector v ON v.profile_id=?1 AND v.owner_kind='relation'
AND v.owner_id=m.id AND v.kind='relation'
WHERE m.deleted=0
AND NOT EXISTS(SELECT 1 FROM chunk_index_job d
WHERE d.profile_id=?1 AND d.owner_kind='relation' AND d.owner_id=m.id
AND d.state='dead')
AND (v.owner_id IS NULL OR v.owner_revision != m.revision)
)
OR EXISTS(
SELECT 1 FROM chunk_vector v
LEFT JOIN taxonomy_relation m ON m.id=v.owner_id
WHERE v.profile_id=?1 AND v.owner_kind='relation'
AND (m.id IS NULL OR m.deleted!=0)
)
OR EXISTS(
SELECT 1 FROM chunk_index_job WHERE profile_id=?1 AND state NOT IN ('done','dead')
)",
[profile.to_string()],
|r| r.get(0),
)
.map_err(sql_error)?;
if invalid {
return Err(MCSError::InvalidParams(
"candidate Full scan has missing or stale vectors/jobs".into(),
));
}
Ok(())
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct AnnGeneration {
pub profile_id: Uuid,
pub durable_generation: i64,
pub published_generation: i64,
pub full_scan_generation: Option<i64>,
}
pub struct AnnGenerationRepository<'a> {
conn: &'a Connection,
}
impl<'a> AnnGenerationRepository<'a> {
pub const fn new(conn: &'a Connection) -> Self {
Self { conn }
}
pub fn get(&self, profile: Uuid) -> Result<AnnGeneration> {
self.conn.query_row("SELECT durable_generation,published_generation,full_scan_generation FROM ann_generation WHERE profile_id=?1", [profile.to_string()], |r| Ok(AnnGeneration {profile_id:profile,durable_generation:r.get(0)?,published_generation:r.get(1)?,full_scan_generation:r.get(2)?})).map_err(sql_error)
}
pub fn verify_full_scan(&self, profile: Uuid) -> Result<()> {
let tx = TxGuard::begin(self.conn)?;
verify_vectors_current(self.conn, profile)?;
let changed = self.conn.execute("UPDATE ann_generation SET full_scan_generation=durable_generation WHERE profile_id=?1", [profile.to_string()]).map_err(sql_error)?;
if changed != 1 {
return Err(MCSError::InvalidParams("unknown ANN profile".into()));
}
tx.commit()
}
/// Call only after building the replacement reader from a consistent read
/// snapshot at this generation. A concurrent durable update rejects publish.
pub fn mark_published(&self, profile: Uuid, generation: i64) -> Result<bool> {
let tx = TxGuard::begin(self.conn)?;
let changed = self.conn.execute("UPDATE ann_generation SET published_generation=?2 WHERE profile_id=?1 AND durable_generation=?2", params![profile.to_string(),generation]).map_err(sql_error)?;
tx.commit()?;
Ok(changed == 1)
}
}
/// Queue one taxonomy subject for one profile. A nil profile means an
/// explicitly held job, in the same way the entity path holds its LegacyCompat
/// fallback. The generation marker resets so a later full-scan verification
/// re-checks this kind from scratch.
pub(crate) fn enqueue_taxonomy(
conn: &Connection,
kind: i64,
id: i64,
revision: i64,
operation: IndexOperation,
profile_id: Uuid,
) -> Result<()> {
// A nil profile is an explicitly held job, mirroring the entity path's
// LegacyCompat fallback: the store has no managed profile to serve it.
let state = if profile_id.is_nil() {
"held"
} else {
"pending"
};
conn.execute("INSERT INTO taxonomy_job(subject_kind,subject_id,profile_id,subject_revision,operation,state) VALUES(?1,?2,?3,?4,?5,?6) ON CONFLICT(subject_kind,subject_id,profile_id) DO UPDATE SET subject_revision=excluded.subject_revision,operation=excluded.operation,state=excluded.state,lease_token=NULL,lease_epoch=lease_epoch+1,lease_until_us=0,attempts=0,next_attempt_us=0,last_error=NULL", params![kind,id,profile_id.to_string(),revision,match operation { IndexOperation::Upsert => "upsert", IndexOperation::Delete => "delete" },state]).map_err(sql_error)?;
// The first subject queued for a managed profile creates the kind's
// generation marker. A commit advances that row and reconciliation
// serves it; without the row, the commit bumps nothing and the kind
// never becomes serveable. A nil profile has nothing to serve.
if !profile_id.is_nil() {
conn.execute(
"INSERT INTO taxonomy_ann_generation(profile_id,subject_kind) VALUES(?1,?2) ON CONFLICT(profile_id,subject_kind) DO NOTHING",
params![profile_id.to_string(), kind],
)
.map_err(sql_error)?;
}
conn.execute(
"UPDATE taxonomy_ann_generation SET full_scan_generation=NULL WHERE profile_id=?1 AND subject_kind=?2",
params![profile_id.to_string(), kind],
)
.map_err(sql_error)?;
Ok(())
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct TaxonomyJob {
pub subject_kind: i64,
pub subject_id: i64,
pub subject_revision: i64,
pub profile_id: Uuid,
pub operation: IndexOperation,
pub lease: Lease,
pub attempts: i64,
}
pub struct TaxonomyJobRepository<'a> {
conn: &'a Connection,
}
impl<'a> TaxonomyJobRepository<'a> {
pub const fn new(conn: &'a Connection) -> Self {
Self { conn }
}
pub fn claim_due(&self, now: i64, duration_us: i64) -> Result<Option<TaxonomyJob>> {
let until = lease_until(now, duration_us)?;
let tx = TxGuard::begin(self.conn)?;
// 0009 purges kind-2 rows in the same startup transaction and commit defends on missing rows; one must never claim.
let row: Option<(i64,i64,String,i64,String,i64,i64)> = self.conn.query_row("SELECT subject_kind,subject_id,profile_id,subject_revision,operation,lease_epoch,attempts FROM taxonomy_job j WHERE ((state='pending' AND next_attempt_us<=?1) OR (state='leased' AND lease_until_us<=?1)) AND subject_kind != 2 AND EXISTS(SELECT 1 FROM index_profile_registry r WHERE r.serving_profile=j.profile_id OR (r.state='Rebuilding' AND r.candidate_profile=j.profile_id)) ORDER BY next_attempt_us,subject_kind,subject_id,profile_id LIMIT 1", [now], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?,r.get(6)?))).optional().map_err(sql_error)?;
let job = row.map(|(kind,id,profile,revision,operation,epoch,attempts)| -> Result<TaxonomyJob> {
let token = Uuid::new_v4();
self.conn.execute("UPDATE taxonomy_job SET state='leased',lease_token=?4,lease_epoch=lease_epoch+1,lease_until_us=?5,attempts=attempts+1 WHERE subject_kind=?1 AND subject_id=?2 AND profile_id=?3", params![kind,id,profile,token.to_string(),until]).map_err(sql_error)?;
Ok(TaxonomyJob { subject_kind:kind,subject_id:id,subject_revision:revision,profile_id:parse_uuid(&profile)?,operation:match operation.as_str() { "upsert"=>IndexOperation::Upsert,"delete"=>IndexOperation::Delete,_=>return Err(MCSError::MemoryError("invalid taxonomy operation".into())) },lease:Lease {token,epoch:epoch+1,until_us:until},attempts:attempts+1 })
}).transpose()?;
tx.commit()?;
Ok(job)
}
pub fn renew(&self, job: &TaxonomyJob, now: i64, duration_us: i64) -> Result<bool> {
let until = lease_until(now, duration_us)?;
let tx = TxGuard::begin(self.conn)?;
let changed = self.conn.execute("UPDATE taxonomy_job SET lease_until_us=?7 WHERE subject_kind=?1 AND subject_id=?2 AND profile_id=?3 AND lease_token=?4 AND lease_epoch=?5 AND state='leased' AND lease_until_us>?6", params![job.subject_kind,job.subject_id,job.profile_id.to_string(),job.lease.token.to_string(),job.lease.epoch,now,until]).map_err(sql_error)?;
tx.commit()?;
Ok(changed == 1)
}
pub fn retry(
&self,
job: &TaxonomyJob,
now: i64,
next_attempt_us: i64,
error: &str,
dead: bool,
) -> Result<bool> {
let tx = TxGuard::begin(self.conn)?;
let changed = self.conn.execute("UPDATE taxonomy_job SET state=?7,next_attempt_us=?8,last_error=?9 WHERE subject_kind=?1 AND subject_id=?2 AND profile_id=?3 AND lease_token=?4 AND lease_epoch=?5 AND state='leased' AND lease_until_us>?6", params![job.subject_kind,job.subject_id,job.profile_id.to_string(),job.lease.token.to_string(),job.lease.epoch,now,if dead {"dead"} else {"pending"},next_attempt_us,error.chars().take(2048).collect::<String>()]).map_err(sql_error)?;
// A dead-lettered subject must not keep a stale vector row: the
// full-scan gate would otherwise publish an outdated embedding. Its
// next write re-enqueues the subject from scratch, mirroring the
// entity dead-letter cleanup.
if changed == 1 && dead {
self.conn
.execute(
"DELETE FROM taxonomy_vector WHERE profile_id=?1 AND subject_kind=?2 AND subject_id=?3",
params![job.profile_id.to_string(), job.subject_kind, job.subject_id],
)
.map_err(sql_error)?;
}
tx.commit()?;
Ok(changed == 1)
}
/// Fenced durable effect and completion are indivisible, mirroring the
/// entity commit. A repeated completion is a no-op.
pub fn commit_vector(
&self,
job: &TaxonomyJob,
now: i64,
vector: Option<&[f32]>,
source: &str,
) -> Result<bool> {
let tx = TxGuard::begin(self.conn)?;
let state: Option<(String,i64)> = self.conn.query_row("SELECT state,lease_until_us FROM taxonomy_job WHERE subject_kind=?1 AND subject_id=?2 AND profile_id=?3 AND subject_revision=?4 AND lease_token=?5 AND lease_epoch=?6", params![job.subject_kind,job.subject_id,job.profile_id.to_string(),job.subject_revision,job.lease.token.to_string(),job.lease.epoch], |r| Ok((r.get(0)?,r.get(1)?))).optional().map_err(sql_error)?;
if matches!(&state,Some((state,_)) if state=="done") {
tx.commit()?;
return Ok(true);
}
if !matches!(state,Some((state,until)) if state=="leased" && until>now) {
return Ok(false);
}
let source_revision: i64 = self
.conn
.query_row(
"SELECT revision FROM type_dict WHERE id=?1 AND kind=?2",
params![job.subject_id, job.subject_kind],
|r| r.get(0),
)
.optional()
.map_err(sql_error)?
.unwrap_or(i64::MAX);
if source_revision != job.subject_revision {
return Ok(false);
}
match (job.operation, vector) {
(IndexOperation::Upsert, Some(vector)) => {
let bytes: Vec<u8> = vector.iter().flat_map(|x| x.to_le_bytes()).collect();
self.conn.execute("INSERT INTO taxonomy_vector VALUES(?1,?2,?3,?4,?5,?6,?7) ON CONFLICT(profile_id,subject_kind,subject_id) DO UPDATE SET subject_revision=excluded.subject_revision,blob=excluded.blob,created_at_us=excluded.created_at_us,source=excluded.source", params![job.profile_id.to_string(),job.subject_kind,job.subject_id,job.subject_revision,bytes,now,source]).map_err(sql_error)?;
}
(IndexOperation::Delete, None) => {
// Kinds 0/1 carry no tombstone and never enqueue deletes; a
// delete against them is a no-op rather than an error, keeping
// the retried-claim path harmless.
}
_ => {
return Err(MCSError::InvalidParams(
"vector payload does not match job operation".into(),
));
}
}
self.conn
.execute(
"UPDATE taxonomy_job SET state='done' WHERE subject_kind=?1 AND subject_id=?2 AND profile_id=?3 AND lease_token=?4 AND lease_epoch=?5",
params![job.subject_kind, job.subject_id, job.profile_id.to_string(), job.lease.token.to_string(), job.lease.epoch],
)
.map_err(sql_error)?;
// The kind generation is the freshness signal for the semantic tier:
// every committed vector retires the kind's snapshot, exactly like the
// entity path retires one per committed entity vector.
self.conn
.execute(
"UPDATE taxonomy_ann_generation SET durable_generation=durable_generation+1,full_scan_generation=NULL WHERE profile_id=?1 AND subject_kind=?2",
params![job.profile_id.to_string(), job.subject_kind],
)
.map_err(sql_error)?;
tx.commit()?;
Ok(true)
}
}
/// Soft full-scan completeness check for one taxonomy kind. It reports whether
/// the kind is missing queued work or carries a stale vector, without failing
/// the caller. Public because the server crate's VectorStore runs it before
/// serving a candidate taxonomy snapshot.
pub fn taxonomy_scan_invalid(conn: &Connection, profile_id: Uuid, kind: i64) -> Result<bool> {
let profile = profile_id.to_string();
let invalid: bool = match kind {
// Kinds 0 and 1 read type_dict members as their source.
0 | 1 => conn.query_row("SELECT EXISTS(SELECT 1 FROM type_dict s WHERE s.kind=?2 AND s.count>0 AND NOT EXISTS(SELECT 1 FROM taxonomy_job j WHERE j.subject_kind=?2 AND j.subject_id=s.id AND j.profile_id=?1 AND j.state!='dead')) OR EXISTS(SELECT 1 FROM taxonomy_vector v JOIN type_dict s ON s.id=v.subject_id WHERE v.profile_id=?1 AND v.subject_kind=?2 AND s.kind=?2 AND s.count>0 AND v.subject_revision!=s.revision)", params![profile,kind], |r| r.get(0)).map_err(sql_error)?,
// Kind 2 derives from the relation chunk rows, so its scan state
// reads the same chunk sources `verify_vectors_current` checks: the
// relation mirror must have a live chunk job and a current vector,
// and no orphaned relation chunk rows may linger.
2 => conn.query_row("SELECT EXISTS(SELECT 1 FROM taxonomy_relation s WHERE s.deleted=0 AND NOT EXISTS(SELECT 1 FROM chunk_index_job j WHERE j.owner_kind='relation' AND j.owner_id=s.id AND j.profile_id=?1 AND j.state!='dead')) OR EXISTS(SELECT 1 FROM chunk_vector v JOIN taxonomy_relation s ON s.id=v.owner_id WHERE v.profile_id=?1 AND v.kind='relation' AND s.deleted=0 AND v.owner_revision!=s.revision)", [profile], |r| r.get(0)).map_err(sql_error)?,
_ => return Err(MCSError::MemoryError("invalid taxonomy subject kind".into())),
};
Ok(invalid)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::schema::initialize_database;
/// In-memory store with every migration applied and one serving profile.
fn fixture() -> (Connection, Uuid) {
let conn = Connection::open_in_memory().unwrap();
initialize_database(&conn).unwrap();
let profile = Uuid::new_v4();
conn.execute(
"UPDATE index_profile_registry SET state='Active',serving_profile=?1 WHERE store_key='default'",
[profile.to_string()],
)
.unwrap();
(conn, profile)
}
fn count(conn: &Connection, table: &str) -> i64 {
conn.query_row(&format!("SELECT count(*) FROM {table}"), [], |r| r.get(0))
.unwrap()
}
fn seed_type(conn: &Connection, id: i64, kind: i64, revision: i64) {
conn.execute(
"INSERT INTO type_dict(id,kind,name,count,revision) VALUES(?1,?2,?3,?4,?5)",
params![id, kind, format!("type{kind}-{id}"), 1, revision],
)
.unwrap();
}
fn seed_relation(conn: &Connection, id: i64, revision: i64, deleted: i64) {
conn.execute(
"INSERT INTO taxonomy_relation(id,from_id,to_id,type_id,revision,deleted) VALUES(?1,?2,?3,?4,?5,?6)",
params![id, id * 100, id * 100 + 1, 2, revision, deleted],
)
.unwrap();
}
#[test]
fn enqueue_after_enqueue_upserts_and_bumps_lease_epoch() {
let (conn, profile) = fixture();
conn.execute(
"INSERT INTO taxonomy_ann_generation(profile_id,subject_kind,durable_generation,full_scan_generation) VALUES(?1,0,5,5)",
[profile.to_string()],
)
.unwrap();
enqueue_taxonomy(&conn, 0, 7, 3, IndexOperation::Upsert, profile).unwrap();
let repo = TaxonomyJobRepository::new(&conn);
let first = repo.claim_due(100, 100).unwrap().unwrap();
assert_eq!(
(first.subject_kind, first.subject_id, first.subject_revision),
(0, 7, 3)
);
enqueue_taxonomy(&conn, 0, 7, 4, IndexOperation::Upsert, profile).unwrap();
let (revision, epoch, token, state): (i64, i64, Option<String>, String) = conn
.query_row(
"SELECT subject_revision,lease_epoch,lease_token,state FROM taxonomy_job WHERE subject_kind=0 AND subject_id=7 AND profile_id=?1",
[profile.to_string()],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
)
.unwrap();
assert_eq!(revision, 4);
assert_eq!(epoch, first.lease.epoch + 1);
assert!(token.is_none());
assert_eq!(state, "pending");
// The superseded lease cannot commit the old revision.
assert!(
!repo
.commit_vector(&first, 101, Some(&[1.0]), "worker")
.unwrap()
);
assert_eq!(count(&conn, "taxonomy_vector"), 0);
let generation: Option<i64> = conn
.query_row(
"SELECT full_scan_generation FROM taxonomy_ann_generation WHERE profile_id=?1 AND subject_kind=0",
[profile.to_string()],
|r| r.get(0),
)
.unwrap();
assert_eq!(generation, None);
}
#[test]
fn claim_picks_the_oldest_due_job() {
let (conn, profile) = fixture();
enqueue_taxonomy(&conn, 1, 3, 1, IndexOperation::Upsert, Uuid::nil()).unwrap();
seed_type(&conn, 1, 0, 1);
enqueue_taxonomy(&conn, 0, 1, 1, IndexOperation::Upsert, profile).unwrap();
seed_type(&conn, 3, 1, 9);
enqueue_taxonomy(&conn, 1, 3, 9, IndexOperation::Upsert, profile).unwrap();
conn.execute(
"UPDATE taxonomy_job SET next_attempt_us=500 WHERE subject_kind=1 AND subject_id=3",
[],
)
.unwrap();
let repo = TaxonomyJobRepository::new(&conn);
let first = repo.claim_due(100, 10).unwrap().unwrap();
assert_eq!((first.subject_kind, first.subject_id), (0, 1));
assert_eq!(first.operation, IndexOperation::Upsert);
assert_eq!((first.lease.epoch, first.lease.until_us), (1, 110));
let (state, attempts): (String, i64) = conn
.query_row(
"SELECT state,attempts FROM taxonomy_job WHERE subject_kind=0 AND subject_id=1 AND profile_id=?1",
[profile.to_string()],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!((state.as_str(), attempts), ("leased", 1));
// The held job is never claimable and the second job is not due yet.
assert!(repo.claim_due(100, 10).unwrap().is_none());
// Complete the first job so its expired lease cannot be re-claimed.
assert!(
repo.commit_vector(&first, 101, Some(&[1.0, 0.0]), "worker")
.unwrap()
);
let second = repo.claim_due(500, 10).unwrap().unwrap();
assert_eq!(
(
second.subject_kind,
second.subject_id,
second.subject_revision
),
(1, 3, 9)
);
assert_eq!(second.lease.until_us, 510);
assert_eq!(count(&conn, "taxonomy_job"), 3);
}
#[test]
fn commit_refuses_after_lease_expiry() {
let (conn, profile) = fixture();
seed_type(&conn, 1, 0, 7);
enqueue_taxonomy(&conn, 0, 1, 7, IndexOperation::Upsert, profile).unwrap();
let repo = TaxonomyJobRepository::new(&conn);
let job = repo.claim_due(100, 10).unwrap().unwrap();
assert!(
!repo
.commit_vector(&job, 111, Some(&[1.0, 0.0]), "worker")
.unwrap()
);
assert_eq!(count(&conn, "taxonomy_vector"), 0);
let state: String = conn
.query_row(
"SELECT state FROM taxonomy_job WHERE subject_kind=0 AND subject_id=1 AND profile_id=?1",
[profile.to_string()],
|r| r.get(0),
)
.unwrap();
assert_eq!(state, "leased");
}
#[test]
fn commit_refuses_on_revision_mismatch_for_claimable_kinds() {
// Kind 2 is retired (Task 5): its rows never claim, so only kinds
// 0 and 1 exercise the revision fence here.
let (conn, profile) = fixture();
seed_type(&conn, 1, 0, 7);
seed_type(&conn, 2, 1, 4);
let repo = TaxonomyJobRepository::new(&conn);
for (kind, id, revision) in [(0, 1, 6), (1, 2, 3)] {
enqueue_taxonomy(&conn, kind, id, revision, IndexOperation::Upsert, profile).unwrap();
let job = repo.claim_due(100 + id, 10).unwrap().unwrap();
assert_eq!((job.subject_kind, job.subject_id), (kind, id));
assert!(
!repo
.commit_vector(&job, 101 + id, Some(&[1.0, 0.0]), "worker")
.unwrap()
);
}
assert_eq!(count(&conn, "taxonomy_vector"), 0);
assert_eq!(count(&conn, "taxonomy_job"), 2);
}
#[test]
fn the_valid_path_succeeds_and_writes_the_vector_row() {
let (conn, profile) = fixture();
seed_type(&conn, 1, 0, 7);
enqueue_taxonomy(&conn, 0, 1, 7, IndexOperation::Upsert, profile).unwrap();
let repo = TaxonomyJobRepository::new(&conn);
let job = repo.claim_due(100, 10).unwrap().unwrap();
assert!(
repo.commit_vector(&job, 105, Some(&[1.0, 0.0]), "worker")
.unwrap()
);
let (kind, revision, blob, created_at, source): (i64, i64, Vec<u8>, i64, String) = conn
.query_row(
"SELECT subject_kind,subject_revision,blob,created_at_us,source FROM taxonomy_vector WHERE profile_id=?1 AND subject_kind=0 AND subject_id=1",
[profile.to_string()],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)),
)
.unwrap();
assert_eq!(kind, 0);
assert_eq!(revision, 7);
assert_eq!(blob, [0, 0, 128, 63, 0, 0, 0, 0]);
assert_eq!(created_at, 105);
assert_eq!(source, "worker");
let state: String = conn
.query_row(
"SELECT state FROM taxonomy_job WHERE subject_kind=0 AND subject_id=1 AND profile_id=?1",
[profile.to_string()],
|r| r.get(0),
)
.unwrap();
assert_eq!(state, "done");
// A repeated completion is a no-op.
assert!(repo.commit_vector(&job, 106, None, "worker").unwrap());
assert_eq!(count(&conn, "taxonomy_vector"), 1);
}
#[test]
fn commit_bumps_the_kind_generation_on_success_only() {
let (conn, profile) = fixture();
conn.execute(
"INSERT INTO taxonomy_ann_generation(profile_id,subject_kind,durable_generation,published_generation,full_scan_generation) VALUES(?1,0,4,-1,4)",
[profile.to_string()],
)
.unwrap();
seed_type(&conn, 1, 0, 7);
enqueue_taxonomy(&conn, 0, 1, 7, IndexOperation::Upsert, profile).unwrap();
let repo = TaxonomyJobRepository::new(&conn);
let job = repo.claim_due(100, 10).unwrap().unwrap();
// A fence-refused commit does not bump the generation. (The enqueue
// already cleared the marker; the durable count is the signal.)
assert!(
!repo
.commit_vector(&job, 111, Some(&[1.0, 0.0]), "worker")
.unwrap()
);
let (durable, full_scan): (i64, Option<i64>) = conn
.query_row(
"SELECT durable_generation,full_scan_generation FROM taxonomy_ann_generation WHERE profile_id=?1 AND subject_kind=0",
[profile.to_string()],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!((durable, full_scan), (4, None));
// A successful commit bumps by one and clears the marker.
assert!(
repo.commit_vector(&job, 105, Some(&[1.0, 0.0]), "worker")
.unwrap()
);
let (durable, full_scan): (i64, Option<i64>) = conn
.query_row(
"SELECT durable_generation,full_scan_generation FROM taxonomy_ann_generation WHERE profile_id=?1 AND subject_kind=0",
[profile.to_string()],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!((durable, full_scan), (5, None));
// A repeated completion is a no-op and does not bump again.
assert!(repo.commit_vector(&job, 106, None, "worker").unwrap());
let durable: i64 = conn
.query_row(
"SELECT durable_generation FROM taxonomy_ann_generation WHERE profile_id=?1 AND subject_kind=0",
[profile.to_string()],
|r| r.get(0),
)
.unwrap();
assert_eq!(durable, 5);
}
#[test]
fn taxonomy_scan_invalid_reports_stale_or_missing_work() {
let (conn, profile) = fixture();
seed_type(&conn, 1, 0, 7);
seed_relation(&conn, 10, 5, 0);
let repo = TaxonomyJobRepository::new(&conn);
// A fully indexed kind is valid.
enqueue_taxonomy(&conn, 0, 1, 7, IndexOperation::Upsert, profile).unwrap();
let job = repo.claim_due(100, 10).unwrap().unwrap();
assert!(
repo.commit_vector(&job, 101, Some(&[1.0, 0.0]), "worker")
.unwrap()
);
assert!(!taxonomy_scan_invalid(&conn, profile, 0).unwrap());
// Kind 2 derives from the relation chunk rows: a live job and a
// current relation chunk make the source current, mirroring the
// entity source but against `chunk_index_job`/`chunk_vector`.
conn.execute(
"INSERT INTO chunk_index_job(profile_id,owner_kind,owner_id,owner_revision,operation,state) VALUES(?1,'relation',10,5,'upsert','done')",
[profile.to_string()],
)
.unwrap();
conn.execute(
"INSERT INTO chunk_vector(profile_id,kind,owner_kind,owner_id,chunk_index,type_id,owner_revision,blob,created_at_us,source) VALUES(?1,'relation','relation',10,0,1,5,X'000000000000803F000000000000803F',1,'old')",
[profile.to_string()],
)
.unwrap();
assert!(!taxonomy_scan_invalid(&conn, profile, 2).unwrap());
// A stale vector is invalid.
conn.execute("UPDATE type_dict SET revision=8 WHERE id=1", [])
.unwrap();
assert!(taxonomy_scan_invalid(&conn, profile, 0).unwrap());
conn.execute("UPDATE type_dict SET revision=7 WHERE id=1", [])
.unwrap();
// A missing job is invalid.
conn.execute(
"DELETE FROM taxonomy_job WHERE subject_kind=0 AND subject_id=1",
[],
)
.unwrap();
assert!(taxonomy_scan_invalid(&conn, profile, 0).unwrap());
assert!(!taxonomy_scan_invalid(&conn, profile, 2).unwrap());
// A pending job counts as queued work even before the vector exists.
seed_relation(&conn, 12, 2, 0);
conn.execute(
"INSERT INTO chunk_index_job(profile_id,owner_kind,owner_id,owner_revision,operation,state) VALUES(?1,'relation',12,2,'upsert','pending')",
[profile.to_string()],
)
.unwrap();
assert!(!taxonomy_scan_invalid(&conn, profile, 2).unwrap());
// A dead job does not count as queued work.
seed_relation(&conn, 11, 3, 0);
conn.execute(
"INSERT INTO chunk_index_job(profile_id,owner_kind,owner_id,owner_revision,operation,state) VALUES(?1,'relation',11,3,'upsert','dead')",
[profile.to_string()],
)
.unwrap();
assert!(taxonomy_scan_invalid(&conn, profile, 2).unwrap());
// A type without members is not a source.
conn.execute("UPDATE type_dict SET count=0 WHERE id=1", [])
.unwrap();
assert!(!taxonomy_scan_invalid(&conn, profile, 0).unwrap());
}
}