Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .github/workflows/test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,12 @@ jobs:
run: cargo fmt --all -- --check
- name: Run clippy
run: cargo clippy --workspace --all-targets --all-features --profile ci -- -D warnings
- name: Run clippy nodedb-client (no features)
run: cargo clippy -p nodedb-client --no-default-features --all-targets --profile ci -- -D warnings
- name: Run clippy nodedb-client (native only)
run: cargo clippy -p nodedb-client --no-default-features --features native --all-targets --profile ci -- -D warnings
- name: Run clippy nodedb-client (remote only)
run: cargo clippy -p nodedb-client --no-default-features --features remote --all-targets --profile ci -- -D warnings
- name: Warm-storage ObjectStore gate
run: bash scripts/ci/check_warm_storage_gate.sh
- name: Format-path .expect() gate
Expand Down
2 changes: 1 addition & 1 deletion docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,7 @@ Admission ordering for requests:
1. Identity & database access (established during authentication)
2. **Tenant** rate/concurrency check → `TENANT_QUOTA_EXCEEDED`
3. **Database** rate/concurrency check → `DATABASE_QUOTA_EXCEEDED`
4. **Global** pressure check → `SERVER_OVERLOAD`
4. **Dispatch** capacity check (tenant in-flight share, database fair share per core, core queue) → `SERVER_OVERLOAD` (code 1403, SQLSTATE `57P03`). Nothing was enqueued, so the client retries with backoff. `nodedb_dispatch_capacity_busy_total{scope}` counts each refusal (pressure). `nodedb_server_overload_responses_total{protocol}` counts each overload answer a client receives (impact). A gap between them is pressure the server absorbs.

Memory reservation flips the order (largest scope first to fail fast): global → database → tenant → engine. Failure at any layer aborts before the next is consulted.

Expand Down
6 changes: 3 additions & 3 deletions nodedb-client/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,9 @@
pub mod capabilities;
pub mod traits;

/// Shared row decoders used by both the trait default impls and the
/// feature-gated clients. Feature-agnostic on purpose — one parser per
/// row shape regardless of which transport delivered the row.
/// Shared row decoders used by the trait default impls and the feature-gated
/// clients. One parser per row shape, whichever transport delivered the row.
/// A decoder used only by the clients compiles under their features.
mod row_decode;

/// Shared SQL escaping helpers (string-literal and identifier quoting).
Expand Down
15 changes: 9 additions & 6 deletions nodedb-client/src/row_decode/mod.rs
Original file line number Diff line number Diff line change
@@ -1,16 +1,19 @@
// SPDX-License-Identifier: Apache-2.0

//! Feature-agnostic decoders that turn `QueryResult` rows back into typed
//! domain structs.
//! Decoders that turn `QueryResult` rows back into typed domain structs.
//!
//! Lives at the crate root so both the (feature-agnostic) `NodeDb` trait
//! default impls and the feature-gated `remote` / `native` clients share
//! one parser per row shape — there is exactly one place to fix when a
//! system-catalog column layout changes.
//! Each row shape has one parser, shared by the `NodeDb` trait default impls
//! and the feature-gated `remote` / `native` clients. A system-catalog column
//! layout change has one place to fix.
//!
//! A decoder compiles under the features of its callers. `search_hit` serves
//! only the client search methods, so it compiles with `native` or `remote`.

pub(crate) mod dropped_collection;
#[cfg(any(feature = "native", feature = "remote"))]
pub(crate) mod search_hit;
pub(crate) mod value;

pub(crate) use dropped_collection::parse_dropped_collection_rows;
#[cfg(any(feature = "native", feature = "remote"))]
pub(crate) use search_hit::{HitSource, decode_search_hits};
39 changes: 39 additions & 0 deletions nodedb-client/src/row_decode/search_hit.rs
Original file line number Diff line number Diff line change
Expand Up @@ -193,4 +193,43 @@ mod tests {
let null_score = vec![vec![Value::String("a".into()), Value::Null]];
assert!(decode_search_hits(&SOURCE, &columns, &null_score).is_err());
}

#[test]
fn a_wrong_cell_type_is_an_error_naming_the_source() {
let columns = names(&["id", "distance"]);
let bool_id = vec![vec![Value::Bool(true), Value::Float(1.0)]];
let err = decode_search_hits(&SOURCE, &columns, &bool_id).expect_err("bool id");
assert!(err.to_string().contains("vector_search 'c'"), "{err}");
assert!(err.to_string().contains("row 0"), "{err}");
let bool_score = vec![
vec![Value::String("a".into()), Value::Float(1.0)],
vec![Value::String("b".into()), Value::Bool(false)],
];
let err = decode_search_hits(&SOURCE, &columns, &bool_score).expect_err("bool score");
assert!(
err.to_string().contains("row 1 ('b') has distance"),
"{err}"
);
}

#[test]
fn the_score_column_comes_from_the_source() {
let text = HitSource {
op: "text_search",
collection: "docs",
score_column: "score",
};
let columns = names(&["id", "distance", "score"]);
let rows = vec![vec![
Value::String("a".into()),
Value::Float(9.0),
Value::Float(0.75),
]];
let hits = decode_search_hits(&text, &columns, &rows).expect("decode");
assert_eq!(hits[0].distance, 0.75);
let err = decode_search_hits(&text, &names(&["id", "distance"]), &rows)
.expect_err("no score column");
assert!(err.to_string().contains("text_search 'docs'"), "{err}");
assert!(err.to_string().contains("'score'"), "{err}");
}
}
30 changes: 24 additions & 6 deletions nodedb-types/src/protocol/text_fields/decode.rs
Original file line number Diff line number Diff line change
Expand Up @@ -267,10 +267,14 @@ impl<'a> zerompk::FromMessagePack<'a> for TextFields {
out.top_k_count = Some(reader.read_u32()?);
}
FID_SCORE_MIN => {
out.score_min = Some(reader.read_binary()?.into_owned());
out.score_min = Some(<crate::value::Value as zerompk::FromMessagePack>::read(
reader,
)?);
}
FID_SCORE_MAX => {
out.score_max = Some(reader.read_binary()?.into_owned());
out.score_max = Some(<crate::value::Value as zerompk::FromMessagePack>::read(
reader,
)?);
}
FID_UPDATES => {
out.updates = Some(Vec::<(String, Vec<u8>)>::read(reader)?);
Expand Down Expand Up @@ -508,8 +512,6 @@ mod tests {
cursor: Some(payload.clone()),
expected: Some(payload.clone()),
new_value: Some(payload.clone()),
score_min: Some(payload.clone()),
score_max: Some(payload.clone()),
filters: Some(payload.clone()),
..Default::default()
};
Expand All @@ -523,11 +525,27 @@ mod tests {
assert_eq!(decoded.cursor.as_deref(), Some(payload.as_slice()));
assert_eq!(decoded.expected.as_deref(), Some(payload.as_slice()));
assert_eq!(decoded.new_value.as_deref(), Some(payload.as_slice()));
assert_eq!(decoded.score_min.as_deref(), Some(payload.as_slice()));
assert_eq!(decoded.score_max.as_deref(), Some(payload.as_slice()));
assert_eq!(decoded.filters.as_deref(), Some(payload.as_slice()));
}

#[test]
fn score_bounds_roundtrip_as_typed_values() {
use crate::value::Value;
for (min, max) in [
(Value::Integer(11), Value::Float(10.5)),
(Value::String("gold".into()), Value::Null),
] {
let tf = TextFields {
score_min: Some(min.clone()),
score_max: Some(max.clone()),
..Default::default()
};
let decoded = roundtrip(&tf);
assert_eq!(decoded.score_min, Some(min));
assert_eq!(decoded.score_max, Some(max));
}
}

#[test]
fn binary_field_encoder_uses_bin_marker_not_fixarray() {
// Wire-shape regression guard. The MessagePack `bin8` marker is
Expand Down
4 changes: 2 additions & 2 deletions nodedb-types/src/protocol/text_fields/encode.rs
Original file line number Diff line number Diff line change
Expand Up @@ -110,8 +110,8 @@ impl zerompk::ToMessagePack for TextFields {
write_opt_field!(writer, FID_WINDOW_START_MS, self.window_start_ms);
write_opt_field!(writer, FID_WINDOW_END_MS, self.window_end_ms);
write_opt_field!(writer, FID_TOP_K_COUNT, self.top_k_count);
write_opt_field!(binary writer, FID_SCORE_MIN, self.score_min);
write_opt_field!(binary writer, FID_SCORE_MAX, self.score_max);
write_opt_field!(writer, FID_SCORE_MIN, self.score_min);
write_opt_field!(writer, FID_SCORE_MAX, self.score_max);
write_opt_field!(writer, FID_UPDATES, self.updates);
write_opt_field!(binary writer, FID_FILTERS, self.filters);
write_opt_field!(writer, FID_VECTOR, self.vector);
Expand Down
10 changes: 6 additions & 4 deletions nodedb-types/src/protocol/text_fields/types/text_fields.rs
Original file line number Diff line number Diff line change
Expand Up @@ -213,12 +213,14 @@ pub struct TextFields {
/// Top-K value for SortedIndexTopK.
#[serde(skip_serializing_if = "Option::is_none")]
pub top_k_count: Option<u32>,
/// Score min for SortedIndexRange (encoded bytes).
/// Inclusive lower score bound for `KvSortedIndexRange`, as a typed value
/// of the index's leading sort column. `None` or `Null` leaves it open.
#[serde(skip_serializing_if = "Option::is_none")]
pub score_min: Option<Vec<u8>>,
/// Score max for SortedIndexRange (encoded bytes).
pub score_min: Option<crate::value::Value>,
/// Inclusive upper score bound for `KvSortedIndexRange`. Same rules as
/// `score_min`.
#[serde(skip_serializing_if = "Option::is_none")]
pub score_max: Option<Vec<u8>>,
pub score_max: Option<crate::value::Value>,

// ── Document advanced ───────────────────────────────────
/// Field-level updates: [(field_name, value_bytes), ...].
Expand Down
105 changes: 105 additions & 0 deletions nodedb-types/src/sync/wire/array/advisory_refused.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
// SPDX-License-Identifier: Apache-2.0

//! Refusal of an advisory array request: an ack or a catch-up request.

use serde::{Deserialize, Serialize};

use super::reject::{ArrayRejectClass, ArrayRejectMsg, ArrayRejectReason};

/// The advisory request Origin refused.
#[derive(
Debug,
Clone,
Copy,
PartialEq,
Eq,
Serialize,
Deserialize,
zerompk::ToMessagePack,
zerompk::FromMessagePack,
)]
#[repr(u8)]
pub enum ArrayAdvisoryKind {
/// An `ArrayAck` (0x95).
Ack = 1,
/// An `ArrayCatchupRequest` (0x97).
CatchupRequest = 2,
}

/// Advisory refusal (server → client, 0x99).
///
/// Names no local op, so it never touches the client's push window.
/// `reason.class()` names the client action, as for `ArrayRejectMsg`.
#[derive(
Debug,
Clone,
PartialEq,
Eq,
Serialize,
Deserialize,
zerompk::ToMessagePack,
zerompk::FromMessagePack,
)]
pub struct ArrayAdvisoryRefusedMsg {
/// Name of the array the request named.
pub array: String,
/// The refused request.
pub request: ArrayAdvisoryKind,
/// Structured reason for the refusal.
pub reason: ArrayRejectReason,
/// Human-readable detail for logs. Clients never parse it.
pub detail: String,
}

impl ArrayAdvisoryRefusedMsg {
/// The refusal of `request` for the reason and detail `reject` carries.
/// The reject's HLC is dropped: an advisory request names no op.
pub fn from_reject(request: ArrayAdvisoryKind, reject: ArrayRejectMsg) -> Self {
Self {
array: reject.array,
request,
reason: reject.reason,
detail: reject.detail,
}
}

/// The action a client takes for this refusal.
pub fn class(&self) -> ArrayRejectClass {
self.reason.class()
}
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn roundtrip_via_msgpack() {
let msg = ArrayAdvisoryRefusedMsg {
array: "scores".into(),
request: ArrayAdvisoryKind::CatchupRequest,
reason: ArrayRejectReason::PermissionDenied,
detail: "no read permission".into(),
};
let encoded = zerompk::to_msgpack_vec(&msg).expect("encode");
let decoded: ArrayAdvisoryRefusedMsg = zerompk::from_msgpack(&encoded).expect("decode");
assert_eq!(msg, decoded);
}

#[test]
fn a_reject_becomes_a_refusal_with_its_reason_and_class() {
let refusal = ArrayAdvisoryRefusedMsg::from_reject(
ArrayAdvisoryKind::Ack,
ArrayRejectMsg {
array: "a".into(),
op_hlc_bytes: [4; 18],
reason: ArrayRejectReason::PermissionDenied,
detail: "denied".into(),
},
);
assert_eq!(refusal.array, "a");
assert_eq!(refusal.request, ArrayAdvisoryKind::Ack);
assert_eq!(refusal.reason, ArrayRejectReason::PermissionDenied);
assert_eq!(refusal.class(), ArrayRejectClass::Permanent);
}
}
6 changes: 5 additions & 1 deletion nodedb-types/src/sync/wire/array/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,19 +9,23 @@
//! payloads via `nodedb-array::sync::op_codec::decode_op` etc.

pub mod ack;
pub mod advisory_refused;
pub mod catchup_request;
pub mod delta;
pub mod delta_batch;
pub mod op_results;
pub mod reject;
pub mod schema;
pub mod snapshot;
pub mod snapshot_chunk;

pub use ack::ArrayAckMsg;
pub use advisory_refused::{ArrayAdvisoryKind, ArrayAdvisoryRefusedMsg};
pub use catchup_request::ArrayCatchupRequestMsg;
pub use delta::ArrayDeltaMsg;
pub use delta_batch::ArrayDeltaBatchMsg;
pub use reject::{ArrayRejectMsg, ArrayRejectReason};
pub use op_results::{ArrayOpOutcome, ArrayOpResult, ArrayOpResultsMsg};
pub use reject::{ArrayRejectClass, ArrayRejectMsg, ArrayRejectReason};
pub use schema::ArraySchemaSyncMsg;
pub use snapshot::ArraySnapshotMsg;
pub use snapshot_chunk::ArraySnapshotChunkMsg;
Loading
Loading