Compare commits
11
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
22d8ad8572 | ||
|
|
7109e67f07 | ||
|
|
52b5a5f909 | ||
|
|
9232662913 | ||
|
|
57d1c5b074 | ||
|
|
ca3abf40f0 | ||
|
|
499e4d7810 | ||
|
|
212cd77cd3 | ||
|
|
7735780807 | ||
|
|
e223f7d327 | ||
|
|
30df055e39 |
@@ -3,11 +3,28 @@
|
||||
# whether a person pushed it or weekly-release.yml created it through the
|
||||
# releases API.
|
||||
#
|
||||
# The image is multi-arch (linux/amd64, linux/arm64) as before, but built in
|
||||
# one buildx run on host1 instead of one native runner per architecture: the
|
||||
# Dockerfile's builder stage runs on the build platform and cross-compiles
|
||||
# with an aarch64 linker, so only the small final stage (apt, setcap) goes
|
||||
# through QEMU for arm64. No digest-joining job is needed.
|
||||
# The image is multi-arch (linux/amd64, linux/arm64), built by two jobs on
|
||||
# the image-build runner rather than one buildx run for both. The Dockerfile's
|
||||
# builder stage runs on the build platform and cross-compiles with an aarch64
|
||||
# linker, so only the small final stage (apt, setcap) goes through QEMU for
|
||||
# arm64 -- but two release builds (LTO, one codegen unit) side by side on one
|
||||
# machine each take twice as long. Production runs amd64, so amd64 goes first
|
||||
# and on its own:
|
||||
# * publish-amd64 pushes :<version>-amd64 and :<version>, a plain amd64
|
||||
# image, as soon as its build is done. A deploy can start from it.
|
||||
# * publish-arm64 then builds arm64, pushes :<version>-arm64, and replaces
|
||||
# :<version> with the two-platform index. :latest moves only here, so it
|
||||
# never names an image without arm64.
|
||||
#
|
||||
# Both jobs use one BuildKit builder, `gitea-builder`, whose container
|
||||
# (buildx_buildkit_gitea-builder0) and state volume stay on the runner's host
|
||||
# between jobs: a job container's `buildx create` finds the existing container
|
||||
# and reuses it and its cache. The dependency build (`cargo chef cook`) is
|
||||
# keyed on the recipe, which only a dependency change alters, so a release
|
||||
# normally compiles just the workspace. Removing that container or its volume
|
||||
# costs the next release a cold build, nothing more. The planner and dependency
|
||||
# layers for the build platform are shared, so arm64 also reuses what amd64
|
||||
# just did where it can.
|
||||
#
|
||||
# Two guards before anything is pushed:
|
||||
# * the tag must be v<brand_version!>. The version is a string in
|
||||
@@ -62,7 +79,7 @@ jobs:
|
||||
echo "version=$V" >> "$GITHUB_OUTPUT"
|
||||
echo "version $V"
|
||||
|
||||
publish:
|
||||
publish-amd64:
|
||||
needs: [version]
|
||||
runs-on: docker
|
||||
container:
|
||||
@@ -81,16 +98,15 @@ jobs:
|
||||
test -n "$REGISTRY" && test -n "$VERSION"
|
||||
test -n "$PACKAGE_TOKEN" || { echo "PACKAGE_TOKEN secret is not set on this repository" >&2; exit 1; }
|
||||
echo "$PACKAGE_TOKEN" | docker login -u jcoffey-dev --password-stdin "$REGISTRY"
|
||||
docker run --privileged --rm tonistiigi/binfmt --install arm64
|
||||
docker buildx create --use --name gitea-builder --driver docker-container || docker buildx use gitea-builder
|
||||
# Attestations off, as before: they add manifests of their own to the
|
||||
# index, and the index should hold the two images and nothing else.
|
||||
# Attestations off, as before: they add manifests of their own, and the
|
||||
# index should hold the two images and nothing else.
|
||||
- run: |
|
||||
docker buildx build \
|
||||
--platform linux/amd64,linux/arm64 \
|
||||
--platform linux/amd64 \
|
||||
--provenance=false --sbom=false \
|
||||
--tag "$IMAGE:$VERSION-amd64" \
|
||||
--tag "$IMAGE:$VERSION" \
|
||||
--tag "$IMAGE:latest" \
|
||||
--push .
|
||||
docker buildx imagetools inspect "$IMAGE:$VERSION"
|
||||
# Gitea keeps a container package on its owner; linking it shows it on
|
||||
@@ -103,11 +119,47 @@ jobs:
|
||||
- if: always()
|
||||
run: docker logout "$REGISTRY" || true
|
||||
|
||||
publish-arm64:
|
||||
needs: [version, publish-amd64]
|
||||
runs-on: docker
|
||||
container:
|
||||
image: docker:28-cli@sha256:625d9431a9f54c5a2bc90f24f0e1c3d55b1349fd857dd85035f98c2c9acbdd4d # 28-cli
|
||||
volumes:
|
||||
- /var/run/docker.sock:/var/run/docker.sock
|
||||
env:
|
||||
DOCKER_BUILDKIT: "1"
|
||||
REGISTRY: ${{ vars.REGISTRY }}
|
||||
IMAGE: ${{ vars.REGISTRY }}/${{ github.repository }}
|
||||
VERSION: ${{ needs.version.outputs.version }}
|
||||
PACKAGE_TOKEN: ${{ secrets.PACKAGE_TOKEN }}
|
||||
steps:
|
||||
- uses: coffey-labs/actions/checkout@fab0c4d45e0162963965f1555df27b7bed5e20ec
|
||||
- run: |
|
||||
echo "$PACKAGE_TOKEN" | docker login -u jcoffey-dev --password-stdin "$REGISTRY"
|
||||
docker run --privileged --rm tonistiigi/binfmt --install arm64
|
||||
docker buildx create --use --name gitea-builder --driver docker-container || docker buildx use gitea-builder
|
||||
# The index is built from the two per-architecture tags rather than from
|
||||
# :<version>, which by now is the amd64 image and would be read as such.
|
||||
- run: |
|
||||
docker buildx build \
|
||||
--platform linux/arm64 \
|
||||
--provenance=false --sbom=false \
|
||||
--tag "$IMAGE:$VERSION-arm64" \
|
||||
--push .
|
||||
docker buildx imagetools create \
|
||||
--tag "$IMAGE:$VERSION" \
|
||||
--tag "$IMAGE:latest" \
|
||||
"$IMAGE:$VERSION-amd64" "$IMAGE:$VERSION-arm64"
|
||||
docker buildx imagetools inspect "$IMAGE:$VERSION"
|
||||
- if: always()
|
||||
run: docker logout "$REGISTRY" || true
|
||||
|
||||
# The weekly release creates its Release (and so the tag) first; a tag
|
||||
# pushed by hand has none. Either way the tag ends up with exactly one
|
||||
# Release, created after the image exists so its pull instructions work.
|
||||
# Release, created once the amd64 image exists so its pull instructions
|
||||
# work; arm64 and the binaries follow.
|
||||
release:
|
||||
needs: [version, publish]
|
||||
needs: [version, publish-amd64]
|
||||
runs-on: light
|
||||
container:
|
||||
image: python:3.13-slim@sha256:8d9d0b8bcf6506481eae4907c18f5e3e7902e629f5f6d684f9e7c32e85e3ddf0 # 3.13-slim
|
||||
@@ -131,7 +183,9 @@ jobs:
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code != 404: raise
|
||||
image = f"{os.environ['REGISTRY']}/{os.environ['REPO']}:{version}"
|
||||
body = (f"Container image: `{image}` (linux/amd64, linux/arm64); also `:latest`.\n\n"
|
||||
body = (f"Container image: `{image}` (linux/amd64, linux/arm64); also `:latest`. "
|
||||
"amd64 is published first; arm64 is added to the same tag when its build "
|
||||
"finishes, and `:latest` moves then.\n\n"
|
||||
"Binaries for a host install are attached: `inbuxa-linux-amd64.tar.gz` and "
|
||||
"`inbuxa-linux-arm64.tar.gz`, with `SHA256SUMS`. Each is the binary out of this "
|
||||
"release's image for that architecture, so it is the same build. The image "
|
||||
@@ -154,7 +208,7 @@ jobs:
|
||||
# `docker create` does not start anything, so pulling an arm64 image on an
|
||||
# amd64 runner and copying a file out of it needs no emulation.
|
||||
binaries:
|
||||
needs: [version, publish, release]
|
||||
needs: [version, publish-arm64, release]
|
||||
runs-on: docker
|
||||
container:
|
||||
image: docker:28-cli@sha256:625d9431a9f54c5a2bc90f24f0e1c3d55b1349fd857dd85035f98c2c9acbdd4d # 28-cli
|
||||
|
||||
@@ -19,6 +19,10 @@ RUN export DEBIAN_FRONTEND=noninteractive && \
|
||||
g++-x86-64-linux-gnu binutils-x86-64-linux-gnu
|
||||
RUN rustup target add "$(cat /target.txt)"
|
||||
COPY --from=planner /recipe.json /recipe.json
|
||||
# inbuxa: [patch.crates-io] points sieve-rs at vendor/, and the recipe only
|
||||
# carries the workspace's own manifests, so cooking the dependencies needs the
|
||||
# vendored crate itself (the context allows it since #27; this puts it here).
|
||||
COPY vendor/ vendor/
|
||||
RUN RUSTFLAGS="$(cat /flags.txt)" cargo chef cook --target "$(cat /target.txt)" --release --no-default-features --features "sqlite postgres mysql rocks s3 redis azure nats" --recipe-path /recipe.json
|
||||
COPY . .
|
||||
RUN RUSTFLAGS="$(cat /flags.txt)" cargo build --target "$(cat /target.txt)" --release -p inbuxa --no-default-features --features "sqlite postgres mysql rocks s3 redis azure nats"
|
||||
|
||||
@@ -23,6 +23,13 @@ use utils::{UnwrapFailure, codec::leb128::Leb128_};
|
||||
|
||||
pub(super) const MAGIC_MARKER: u8 = 123;
|
||||
|
||||
// inbuxa: blobs kept under a fixed name instead of a content hash. Nothing
|
||||
// links to them, so the export names them outright.
|
||||
const NAMED_BLOBS: &[&[u8]] = &[
|
||||
crate::manager::SPAM_CLASSIFIER_KEY,
|
||||
crate::manager::SPAM_TRAINER_KEY,
|
||||
];
|
||||
|
||||
#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
|
||||
pub(super) enum Family {
|
||||
Data = 0,
|
||||
@@ -143,15 +150,21 @@ impl Core {
|
||||
.await
|
||||
.failed("Failed to iterate over data store");
|
||||
|
||||
for hash in blobs {
|
||||
// inbuxa: the trained spam classifier and its trainer state are
|
||||
// blobs stored under fixed names with no blob link, so the walk
|
||||
// over links above never reaches them.
|
||||
let named = NAMED_BLOBS.iter().map(|key| key.to_vec());
|
||||
for key in blobs
|
||||
.into_iter()
|
||||
.map(|hash| hash.as_slice().to_vec())
|
||||
.chain(named)
|
||||
{
|
||||
if let Some(blob) = blob_store
|
||||
.get_blob(hash.as_slice(), 0..usize::MAX)
|
||||
.get_blob(&key, 0..usize::MAX)
|
||||
.await
|
||||
.failed("Failed to get blob")
|
||||
{
|
||||
writer
|
||||
.send((hash.as_slice().to_vec(), blob))
|
||||
.failed("Failed to send key");
|
||||
writer.send((key, blob)).failed("Failed to send key");
|
||||
}
|
||||
}
|
||||
}),
|
||||
@@ -323,7 +336,13 @@ impl Family {
|
||||
SUBSPACE_REGISTRY_IDX,
|
||||
SUBSPACE_REGISTRY_PK,
|
||||
SUBSPACE_DIRECTORY,
|
||||
store::SUBSPACE_INBUXA, // inbuxa: masked email
|
||||
// inbuxa: registry objects the upstream list left out, so an
|
||||
// export dropped them: archived items (undelete) and spam
|
||||
// training samples. Their indexes and id counters already
|
||||
// travel in this family and in `data`, so they ride along.
|
||||
SUBSPACE_DELETED_ITEMS,
|
||||
SUBSPACE_SPAM_SAMPLES,
|
||||
store::SUBSPACE_INBUXA, // inbuxa: the fork's own data (masked email, undelete, policies)
|
||||
],
|
||||
Family::Changelog => &[SUBSPACE_LOGS],
|
||||
Family::Queue => &[SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUEUE_EVENT],
|
||||
|
||||
@@ -54,6 +54,13 @@ Options:
|
||||
-o, --console Open the store console
|
||||
-h, --help Print help
|
||||
-V, --version Print version
|
||||
|
||||
An export holds everything in the data and blob stores except short-lived
|
||||
in-memory state (rate limits, locks, greylisting) and the full-text search
|
||||
index, which belongs to one search backend. An import into an empty store
|
||||
queues the index to be rebuilt when the server next starts. EXPORT_TYPES
|
||||
limits an export to some of: data, registry, blob, changelog, queue, report,
|
||||
telemetry, tasks.
|
||||
"#
|
||||
);
|
||||
|
||||
@@ -256,10 +263,10 @@ impl BootManager {
|
||||
telemetry.enable();
|
||||
|
||||
// Parse settings and restore
|
||||
Box::pin(Core::parse(&mut bootstrap, storage))
|
||||
.await
|
||||
.restore(path)
|
||||
.await;
|
||||
let core = Box::pin(Core::parse(&mut bootstrap, storage)).await;
|
||||
let imported = core.restore(path).await;
|
||||
// inbuxa: the search index isn't exported; rebuild it
|
||||
core.queue_reindex(&imported).await;
|
||||
std::process::exit(0);
|
||||
}
|
||||
StoreOp::Console => {
|
||||
|
||||
@@ -9,15 +9,22 @@
|
||||
use super::backup::MAGIC_MARKER;
|
||||
use crate::{Core, DATABASE_SCHEMA_VERSION};
|
||||
use lz4_flex::frame::FrameDecoder;
|
||||
use registry::schema::enums::CompressionAlgo;
|
||||
use registry::{
|
||||
schema::{
|
||||
enums::{CompressionAlgo, TaskStoreMaintenanceType},
|
||||
structs::{Task, TaskStatus, TaskStoreMaintenance},
|
||||
},
|
||||
types::EnumImpl,
|
||||
};
|
||||
use std::{
|
||||
fs::File,
|
||||
io::{BufReader, ErrorKind, Read},
|
||||
path::{Path, PathBuf},
|
||||
};
|
||||
use store::{
|
||||
BlobStore, IterateParams, SUBSPACE_BLOBS, SUBSPACE_COUNTER, SUBSPACE_INDEXES, SUBSPACE_QUOTA,
|
||||
SUBSPACE_REGISTRY_PK, Store, U32_LEN,
|
||||
BlobStore, IterateParams, SUBSPACE_BLOBS, SUBSPACE_COUNTER, SUBSPACE_INDEXES,
|
||||
SUBSPACE_PROPERTY, SUBSPACE_QUOTA, SUBSPACE_REGISTRY_PK, SUBSPACE_TELEMETRY_SPAN, Store,
|
||||
U32_LEN,
|
||||
write::{
|
||||
AnyClass, AnyKey, BatchBuilder, ValueClass,
|
||||
key::{DeserializeBigEndian, is_node_id_key},
|
||||
@@ -27,7 +34,9 @@ use types::{collection::Collection, field::Field};
|
||||
use utils::{UnwrapFailure, failed};
|
||||
|
||||
impl Core {
|
||||
pub async fn restore(&self, src: PathBuf) {
|
||||
/// Imports an export into an empty store and returns the subspaces it
|
||||
/// wrote. inbuxa: the caller hands them to [`Core::queue_reindex`].
|
||||
pub async fn restore(&self, src: PathBuf) -> Vec<u8> {
|
||||
// Backup the core
|
||||
let paths = if src.is_dir() {
|
||||
let mut paths = Vec::new();
|
||||
@@ -64,6 +73,13 @@ impl Core {
|
||||
std::process::exit(1);
|
||||
}
|
||||
|
||||
let mut imported = paths
|
||||
.iter()
|
||||
.map(|path| KeyValueReader::new(path).subspace)
|
||||
.collect::<Vec<_>>();
|
||||
imported.sort_unstable();
|
||||
imported.dedup();
|
||||
|
||||
let mut tasks = Vec::new();
|
||||
for path in paths {
|
||||
let storage = self.storage.clone();
|
||||
@@ -76,6 +92,54 @@ impl Core {
|
||||
for task in tasks {
|
||||
task.await.failed("Failed to wait for task");
|
||||
}
|
||||
|
||||
imported
|
||||
}
|
||||
|
||||
/// inbuxa: an export never carries the full-text index. It is built by
|
||||
/// and for one search backend (the SQL stores index into their own
|
||||
/// tables, the key-value stores into a subspace, external engines keep it
|
||||
/// themselves), so it would be wrong or unreadable after a move to
|
||||
/// another one. Instead, an import queues the same reindex tasks an
|
||||
/// administrator can queue by hand (`reindexAccounts` and
|
||||
/// `reindexTelemetry` store maintenance), and the server rebuilds the
|
||||
/// index for whatever search store it is configured with once it starts.
|
||||
pub async fn queue_reindex(&self, imported: &[u8]) -> Vec<TaskStoreMaintenanceType> {
|
||||
let mut queued = Vec::new();
|
||||
if imported.contains(&SUBSPACE_PROPERTY) {
|
||||
queued.push(TaskStoreMaintenanceType::ReindexAccounts);
|
||||
}
|
||||
if imported.contains(&SUBSPACE_TELEMETRY_SPAN) {
|
||||
queued.push(TaskStoreMaintenanceType::ReindexTelemetry);
|
||||
}
|
||||
if queued.is_empty() {
|
||||
return queued;
|
||||
}
|
||||
|
||||
let mut batch = BatchBuilder::new();
|
||||
for maintenance_type in &queued {
|
||||
batch.schedule_task(Task::StoreMaintenance(TaskStoreMaintenance {
|
||||
maintenance_type: *maintenance_type,
|
||||
status: TaskStatus::now(),
|
||||
shard_index: None,
|
||||
}));
|
||||
}
|
||||
self.storage
|
||||
.data
|
||||
.write(batch.build_all())
|
||||
.await
|
||||
.failed("Failed to queue the reindex tasks");
|
||||
|
||||
println!(
|
||||
"Queued {} to rebuild the search index; it runs when the server starts.",
|
||||
queued
|
||||
.iter()
|
||||
.map(|t| t.as_str())
|
||||
.collect::<Vec<_>>()
|
||||
.join(" and ")
|
||||
);
|
||||
|
||||
queued
|
||||
}
|
||||
}
|
||||
|
||||
@@ -125,17 +189,22 @@ async fn restore_file(store: Store, blob_store: BlobStore, path: &Path) {
|
||||
}
|
||||
SUBSPACE_COUNTER | SUBSPACE_QUOTA => {
|
||||
while let Some((key, value)) = reader.next() {
|
||||
batch.add(
|
||||
ValueClass::Any(AnyClass {
|
||||
let class = ValueClass::Any(AnyClass {
|
||||
subspace: reader.subspace,
|
||||
key,
|
||||
}),
|
||||
u64::from_le_bytes(
|
||||
});
|
||||
let value = u64::from_le_bytes(
|
||||
value
|
||||
.try_into()
|
||||
.expect("Failed to deserialize counter/quota"),
|
||||
) as i64,
|
||||
);
|
||||
) as i64;
|
||||
// inbuxa: the SQL stores add a negative amount with an UPDATE,
|
||||
// which does nothing to a row that isn't there yet, so a
|
||||
// negative counter vanished on import. Create the row first.
|
||||
if value < 0 {
|
||||
batch.add(class.clone(), 0);
|
||||
}
|
||||
batch.add(class, value);
|
||||
if batch.is_large_batch() {
|
||||
store
|
||||
.write(batch.build_all())
|
||||
|
||||
@@ -427,9 +427,24 @@ pub(crate) async fn trace_query(
|
||||
}
|
||||
None => false,
|
||||
},
|
||||
Property::QueueId => match value.as_str() {
|
||||
// The queue id column is an integer on every search backend, and
|
||||
// holds a trace's first queue id; the keywords carry all of them
|
||||
Property::QueueId => match value
|
||||
.as_str()
|
||||
.and_then(|v| v.trim().parse::<u64>().ok())
|
||||
.or_else(|| value.as_u64())
|
||||
{
|
||||
Some(queue_id) => {
|
||||
search.push(SearchFilter::eq(TracingSearchField::QueueId, queue_id.to_string()));
|
||||
search.extend([
|
||||
SearchFilter::Or,
|
||||
SearchFilter::eq(TracingSearchField::QueueId, queue_id),
|
||||
SearchFilter::has_text(
|
||||
TracingSearchField::Keywords,
|
||||
queue_id.to_string(),
|
||||
nlp::language::Language::None,
|
||||
),
|
||||
SearchFilter::End,
|
||||
]);
|
||||
true
|
||||
}
|
||||
None => false,
|
||||
|
||||
@@ -26,7 +26,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
|
||||
};
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut retry_count = 0;
|
||||
let mut retry_count: u32 = 0;
|
||||
|
||||
trc::event!(Cluster(ClusterEvent::SubscriberStart));
|
||||
|
||||
@@ -53,7 +53,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
|
||||
);
|
||||
|
||||
match tokio::time::timeout(
|
||||
Duration::from_secs(1 << retry_count.max(6)),
|
||||
subscribe_retry_delay(retry_count),
|
||||
shutdown_rx.changed(),
|
||||
)
|
||||
.await
|
||||
@@ -62,7 +62,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
|
||||
break;
|
||||
}
|
||||
Err(_) => {
|
||||
retry_count += 1;
|
||||
retry_count = retry_count.saturating_add(1);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
@@ -234,6 +234,11 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
|
||||
});
|
||||
}
|
||||
|
||||
/// Delay before the next subscribe attempt: 1 s, 2 s, 4 s ... capped at 64 s.
|
||||
fn subscribe_retry_delay(retry_count: u32) -> Duration {
|
||||
Duration::from_secs(1u64 << retry_count.min(6))
|
||||
}
|
||||
|
||||
fn log_event(event: &BroadcastEvent) -> trc::Value {
|
||||
match event {
|
||||
BroadcastEvent::PushNotification(notification) => match notification {
|
||||
@@ -296,3 +301,19 @@ fn log_event(event: &BroadcastEvent) -> trc::Value {
|
||||
BroadcastEvent::QueueRefresh => "QueueRefresh".into(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::subscribe_retry_delay;
|
||||
use std::time::Duration;
|
||||
|
||||
#[test]
|
||||
fn subscribe_retry_backoff_grows_then_caps() {
|
||||
let schedule: Vec<u64> = (0..10)
|
||||
.map(|n| subscribe_retry_delay(n).as_secs())
|
||||
.collect();
|
||||
assert_eq!(schedule, vec![1, 2, 4, 8, 16, 32, 64, 64, 64, 64]);
|
||||
// No shift overflow at the top of the range.
|
||||
assert_eq!(subscribe_retry_delay(u32::MAX), Duration::from_secs(64));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -567,20 +567,14 @@ async fn build_contact_document(
|
||||
}
|
||||
|
||||
|
||||
// inbuxa: MON-16: a trace's search document, when trace search is on:
|
||||
// its event types, queue ids, and addresses, their domains, hosts, IPs,
|
||||
// message ids and account names as keywords
|
||||
// inbuxa: MON-16: a trace's search document, when trace search is on
|
||||
async fn build_tracing_span_document(
|
||||
server: &Server,
|
||||
span_id: u64,
|
||||
) -> trc::Result<Option<IndexDocument>> {
|
||||
use common::telemetry::tracers::store::MaybeTrace;
|
||||
use registry::schema::{enums::SearchTracingField, structs::Search};
|
||||
use store::{
|
||||
search::TracingSearchField,
|
||||
write::{TelemetryClass, ValueClass},
|
||||
};
|
||||
use trc::Key;
|
||||
use registry::schema::structs::Search;
|
||||
use store::write::{TelemetryClass, ValueClass};
|
||||
|
||||
let settings = server
|
||||
.registry()
|
||||
@@ -590,7 +584,6 @@ async fn build_tracing_span_document(
|
||||
if !settings.index_telemetry {
|
||||
return Ok(None);
|
||||
}
|
||||
let wants = |field: SearchTracingField| settings.index_tracing_fields.iter().any(|f| *f == field);
|
||||
let Some(MaybeTrace(Some(trace))) = server
|
||||
.tracing_store()
|
||||
.get_value::<MaybeTrace>(ValueKey::from(ValueClass::Telemetry(TelemetryClass::Span(
|
||||
@@ -601,23 +594,67 @@ async fn build_tracing_span_document(
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let mut document = IndexDocument::new(SearchIndex::Tracing).with_id(span_id);
|
||||
let mut seen = store::ahash::AHashSet::new();
|
||||
for event in trace.events.iter() {
|
||||
if wants(SearchTracingField::EventType) && seen.insert(event.event.as_str().to_string()) {
|
||||
document.index_keyword(TracingSearchField::EventType, event.event.as_str());
|
||||
Ok(Some(trace_search_document(
|
||||
span_id,
|
||||
&trace,
|
||||
&settings
|
||||
.index_tracing_fields
|
||||
.iter()
|
||||
.copied()
|
||||
.collect::<Vec<_>>(),
|
||||
)))
|
||||
}
|
||||
|
||||
/// inbuxa: MON-16: the search document for a stored trace.
|
||||
///
|
||||
/// The event type and queue id columns are integers on every search backend
|
||||
/// (BIGINT on PostgreSQL and MySQL, long on Elasticsearch), and each holds a
|
||||
/// single value per trace: the event type is the trace's opening event, the
|
||||
/// one `x:Trace/query` filters on, and the queue id is the first queue id the
|
||||
/// trace mentions. Every queue id also goes into the keywords, so a session
|
||||
/// that queued several messages is found by any of them.
|
||||
pub fn trace_search_document(
|
||||
span_id: u64,
|
||||
trace: ®istry::schema::structs::Trace,
|
||||
fields: &[registry::schema::enums::SearchTracingField],
|
||||
) -> IndexDocument {
|
||||
use registry::schema::{enums::SearchTracingField, structs::TraceValue};
|
||||
use store::search::TracingSearchField;
|
||||
use trc::Key;
|
||||
|
||||
let wants = |field: SearchTracingField| fields.contains(&field);
|
||||
let mut document = IndexDocument::new(SearchIndex::Tracing).with_id(span_id);
|
||||
if wants(SearchTracingField::EventType)
|
||||
&& let Some(first) = trace.events.iter().next()
|
||||
{
|
||||
document.index_unsigned(TracingSearchField::EventType, first.event.to_id() as u64);
|
||||
}
|
||||
|
||||
let mut seen = store::ahash::AHashSet::new();
|
||||
let mut queue_id_indexed = false;
|
||||
for event in trace.events.iter() {
|
||||
for kv in event.key_values.iter() {
|
||||
let text = match &kv.value {
|
||||
registry::schema::structs::TraceValue::String(v) => v.value.clone(),
|
||||
registry::schema::structs::TraceValue::UnsignedInt(v) => v.value.to_string(),
|
||||
registry::schema::structs::TraceValue::IpAddr(v) => v.value.to_string(),
|
||||
TraceValue::String(v) => v.value.clone(),
|
||||
TraceValue::UnsignedInt(v) => v.value.to_string(),
|
||||
TraceValue::IpAddr(v) => v.value.to_string(),
|
||||
_ => continue,
|
||||
};
|
||||
match kv.key {
|
||||
Key::QueueId if wants(SearchTracingField::QueueId) => {
|
||||
if seen.insert(format!("q:{text}")) {
|
||||
document.index_keyword(TracingSearchField::QueueId, &text);
|
||||
Key::QueueId => {
|
||||
let Ok(queue_id) = text.parse::<u64>() else {
|
||||
continue;
|
||||
};
|
||||
if wants(SearchTracingField::QueueId) && !queue_id_indexed {
|
||||
document.index_unsigned(TracingSearchField::QueueId, queue_id);
|
||||
queue_id_indexed = true;
|
||||
}
|
||||
if wants(SearchTracingField::Keywords) && seen.insert(format!("k:{text}")) {
|
||||
document.index_text(
|
||||
TracingSearchField::Keywords,
|
||||
&text,
|
||||
nlp::language::Language::None,
|
||||
);
|
||||
}
|
||||
}
|
||||
Key::From
|
||||
@@ -648,7 +685,7 @@ async fn build_tracing_span_document(
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(Some(document))
|
||||
document
|
||||
}
|
||||
|
||||
// inbuxa: UD-1, UD-4: archives a deleted file, event or contact noted at
|
||||
|
||||
@@ -81,7 +81,7 @@ fn legacy_setting(name: &str, is_set: impl Fn(&str) -> bool) -> Option<String> {
|
||||
#[macro_export]
|
||||
macro_rules! brand_version {
|
||||
() => {
|
||||
"2026.9.24.2"
|
||||
"2026.9.24.3"
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -212,10 +212,15 @@ unchanged.
|
||||
- **MON-16.** With `indexTelemetry` on, storing a trace schedules an
|
||||
`IndexTrace` task. The task builds one document for `SearchIndex::Tracing`
|
||||
with the fields named in `indexTracingFields`:
|
||||
- `eventType`: every event type in the trace;
|
||||
- `queueId`: every `queueId` value;
|
||||
- `eventType`: the trace's opening event, as its numeric id;
|
||||
- `queueId`: the first `queueId` value, as an integer;
|
||||
- `keywords`: every address in `from` and `to`, each address's domain, every
|
||||
`domain`, `hostname`, `remoteIp`, `messageId` and `accountName` value.
|
||||
`domain`, `hostname`, `remoteIp`, `messageId` and `accountName` value,
|
||||
and every `queueId` value.
|
||||
The event type and queue id are single integer columns on every search
|
||||
backend (BIGINT on PostgreSQL and MySQL), so the `queueId` filter matches
|
||||
the column or any queue id in the keywords, and a session that queued
|
||||
several messages is found by each of them.
|
||||
So searching `example.org` finds every trace to or from that domain, as the
|
||||
upstream suite expects. With `indexTelemetry` off nothing is indexed, and
|
||||
the `text` and `queueId` filters are refused (see "Interfaces").
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::utils::{
|
||||
@@ -9,14 +11,22 @@ use crate::utils::{
|
||||
server::TestServer,
|
||||
temp_dir::TempDir,
|
||||
};
|
||||
use ::registry::schema::enums::CompressionAlgo;
|
||||
use ::registry::schema::{
|
||||
enums::{CompressionAlgo, TaskStoreMaintenanceType},
|
||||
prelude::ObjectType,
|
||||
structs::Task,
|
||||
};
|
||||
use ahash::AHashSet;
|
||||
use common::{DATABASE_SCHEMA_VERSION, manager::backup::BackupParams};
|
||||
use common::{
|
||||
DATABASE_SCHEMA_VERSION,
|
||||
manager::{SPAM_CLASSIFIER_KEY, SPAM_TRAINER_KEY, backup::BackupParams},
|
||||
};
|
||||
use store::{
|
||||
rand,
|
||||
write::{
|
||||
AnyClass, AnyKey, BatchBuilder, BlobLink, BlobOp, Operation, QueueClass, QueueEvent,
|
||||
RegistryClass, ValueClass, key::KeySerializer,
|
||||
RegistryClass, TaskQueueClass, ValueClass,
|
||||
key::{DeserializeBigEndian, KeySerializer},
|
||||
},
|
||||
*,
|
||||
};
|
||||
@@ -167,6 +177,50 @@ pub async fn test(test: &TestServer) {
|
||||
}
|
||||
db.write(batch.build_all()).await.unwrap();
|
||||
|
||||
// inbuxa: registry objects kept outside the registry subspace (archived
|
||||
// items for undelete, spam training samples, directory entries) and the
|
||||
// fork's own subspace. Exports used to leave the first two behind.
|
||||
println!("Creating archived items, spam samples and fork data...");
|
||||
let mut batch = BatchBuilder::new();
|
||||
for item_id in [1u64, 2, 3] {
|
||||
for object in [
|
||||
ObjectType::ArchivedItem,
|
||||
ObjectType::SpamTrainingSample,
|
||||
ObjectType::Account,
|
||||
] {
|
||||
batch.set(
|
||||
ValueClass::Registry(RegistryClass::Item {
|
||||
object_id: object as u16,
|
||||
item_id,
|
||||
}),
|
||||
random_bytes(item_id as usize * 64),
|
||||
);
|
||||
}
|
||||
batch.set(
|
||||
ValueClass::Any(AnyClass {
|
||||
subspace: SUBSPACE_INBUXA,
|
||||
key: [b'U', b'x']
|
||||
.into_iter()
|
||||
.chain(item_id.to_be_bytes())
|
||||
.collect(),
|
||||
}),
|
||||
random_bytes(32),
|
||||
);
|
||||
}
|
||||
db.write(batch.build_all()).await.unwrap();
|
||||
|
||||
// inbuxa: the trained spam classifier lives in blobs with fixed names
|
||||
let mut named_blobs = Vec::new();
|
||||
for key in [SPAM_CLASSIFIER_KEY, SPAM_TRAINER_KEY] {
|
||||
let data = random_bytes(4096);
|
||||
test.server
|
||||
.blob_store()
|
||||
.put_blob(key, &data, CompressionAlgo::Lz4)
|
||||
.await
|
||||
.unwrap();
|
||||
named_blobs.push((key, data));
|
||||
}
|
||||
|
||||
// Create directory data
|
||||
println!("Creating directory data...");
|
||||
let mut batch = BatchBuilder::new();
|
||||
@@ -185,6 +239,17 @@ pub async fn test(test: &TestServer) {
|
||||
println!("Calculating store hash...");
|
||||
let snapshot = Snapshot::new(&db).await;
|
||||
assert!(!snapshot.keys.is_empty(), "Store hash counts are empty",);
|
||||
for subspace in [
|
||||
SUBSPACE_DELETED_ITEMS,
|
||||
SUBSPACE_SPAM_SAMPLES,
|
||||
SUBSPACE_INBUXA,
|
||||
] {
|
||||
assert!(
|
||||
snapshot.keys.iter().any(|k| k.subspace == subspace),
|
||||
"No test data in subspace {}",
|
||||
char::from(subspace)
|
||||
);
|
||||
}
|
||||
|
||||
// Export store
|
||||
println!("Exporting store...");
|
||||
@@ -210,22 +275,188 @@ pub async fn test(test: &TestServer) {
|
||||
.finalize(),
|
||||
);
|
||||
db.write(batch.build_all()).await.unwrap();
|
||||
test.server.core.restore(temp_dir.path.clone()).await;
|
||||
for (key, _) in &named_blobs {
|
||||
test.server.blob_store().delete_blob(key).await.unwrap();
|
||||
}
|
||||
let imported = test.server.core.restore(temp_dir.path.clone()).await;
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.clear(ValueClass::NodeId(0));
|
||||
db.write(batch.build_all()).await.unwrap();
|
||||
for subspace in [
|
||||
SUBSPACE_DELETED_ITEMS,
|
||||
SUBSPACE_SPAM_SAMPLES,
|
||||
SUBSPACE_INBUXA,
|
||||
] {
|
||||
assert!(
|
||||
imported.contains(&subspace),
|
||||
"Subspace {} was not exported",
|
||||
char::from(subspace)
|
||||
);
|
||||
}
|
||||
|
||||
// Verify hash
|
||||
print!("Verifying store hash...");
|
||||
snapshot.assert_is_eq(&Snapshot::new(&db).await);
|
||||
assert_named_blobs(test.server.blob_store(), &named_blobs).await;
|
||||
println!(" GREAT SUCCESS!");
|
||||
|
||||
// inbuxa: import the same export into a fresh store of another backend,
|
||||
// the way a move from one database to another does it
|
||||
#[cfg(all(feature = "rocks", feature = "sqlite"))]
|
||||
cross_backend(test, &db, &temp_dir, &named_blobs).await;
|
||||
|
||||
// Destroy store
|
||||
for (key, _) in &named_blobs {
|
||||
test.server.blob_store().delete_blob(key).await.unwrap();
|
||||
}
|
||||
store_destroy(&db).await;
|
||||
store_assert_is_empty(&db, db.clone().into(), true).await;
|
||||
temp_dir.delete();
|
||||
}
|
||||
|
||||
#[cfg(all(feature = "rocks", feature = "sqlite"))]
|
||||
async fn cross_backend(
|
||||
test: &TestServer,
|
||||
source: &Store,
|
||||
export: &TempDir,
|
||||
named_blobs: &[(&[u8], Vec<u8>)],
|
||||
) {
|
||||
let source_type = std::env::var("STORE").unwrap();
|
||||
let target_type = if source_type.eq_ignore_ascii_case("sqlite") {
|
||||
"RocksDb"
|
||||
} else {
|
||||
"Sqlite"
|
||||
};
|
||||
println!("Importing the export into a fresh {target_type} store...");
|
||||
|
||||
let target_dir = TempDir::new("art_vandelay_cross_backend", true);
|
||||
let target = Store::build(
|
||||
crate::utils::storage::build_data_store(target_type, &target_dir.path.to_string_lossy())
|
||||
.await,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
target.create_tables().await.unwrap();
|
||||
store_destroy(&target).await;
|
||||
|
||||
let mut core = test.server.core.as_ref().clone();
|
||||
core.storage.data = target.clone();
|
||||
core.storage.blob = target.clone().into();
|
||||
let imported = core.restore(export.path.clone()).await;
|
||||
|
||||
// Counters are stored differently by the SQL and key-value backends, so
|
||||
// compare their keys here and their values through the counter API.
|
||||
print!("Verifying {target_type} store hash...");
|
||||
Snapshot::new_portable(source)
|
||||
.await
|
||||
.assert_is_eq(&Snapshot::new_portable(&target).await);
|
||||
for subspace in [SUBSPACE_COUNTER, SUBSPACE_QUOTA] {
|
||||
let mut keys = Vec::new();
|
||||
source
|
||||
.iterate(
|
||||
IterateParams::new(
|
||||
AnyKey {
|
||||
subspace,
|
||||
key: vec![0u8],
|
||||
},
|
||||
AnyKey {
|
||||
subspace,
|
||||
key: vec![u8::MAX; 10],
|
||||
},
|
||||
)
|
||||
.no_values(),
|
||||
|key, _| {
|
||||
keys.push(key.to_vec());
|
||||
Ok(true)
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
for key in keys {
|
||||
let class = || {
|
||||
ValueClass::Any(AnyClass {
|
||||
subspace,
|
||||
key: key.clone(),
|
||||
})
|
||||
};
|
||||
assert_eq!(
|
||||
source.get_counter(class()).await.unwrap(),
|
||||
target.get_counter(class()).await.unwrap(),
|
||||
"Counter mismatch in {} for {key:?}",
|
||||
char::from(subspace)
|
||||
);
|
||||
}
|
||||
}
|
||||
assert_named_blobs(&core.storage.blob, named_blobs).await;
|
||||
println!(" GREAT SUCCESS!");
|
||||
|
||||
// The search index isn't exported; the import queues its rebuild
|
||||
let queued = core.queue_reindex(&imported).await;
|
||||
let expected = [
|
||||
TaskStoreMaintenanceType::ReindexAccounts,
|
||||
TaskStoreMaintenanceType::ReindexTelemetry,
|
||||
];
|
||||
assert_eq!(queued, expected);
|
||||
let mut task_ids = Vec::new();
|
||||
target
|
||||
.iterate(
|
||||
IterateParams::new(
|
||||
AnyKey {
|
||||
subspace: SUBSPACE_TASK_QUEUE,
|
||||
key: vec![0u8],
|
||||
},
|
||||
AnyKey {
|
||||
subspace: SUBSPACE_TASK_QUEUE,
|
||||
key: vec![u8::MAX; 20],
|
||||
},
|
||||
)
|
||||
.no_values(),
|
||||
|key, _| {
|
||||
if key.deserialize_be_u64(0)? == 0 {
|
||||
task_ids.push(key.deserialize_be_u64(U64_LEN)?);
|
||||
}
|
||||
Ok(true)
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let mut found = Vec::new();
|
||||
for id in task_ids {
|
||||
match target
|
||||
.get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
|
||||
TaskQueueClass::Task { id },
|
||||
)))
|
||||
.await
|
||||
.unwrap()
|
||||
{
|
||||
Some(Task::StoreMaintenance(task)) => found.push(task.maintenance_type),
|
||||
other => panic!("Unexpected task {other:?}"),
|
||||
}
|
||||
}
|
||||
found.sort_by_key(|t| *t as u16);
|
||||
assert_eq!(found, expected, "Queued tasks don't match");
|
||||
|
||||
store_destroy(&target).await;
|
||||
drop(core);
|
||||
drop(target);
|
||||
target_dir.delete();
|
||||
}
|
||||
|
||||
async fn assert_named_blobs(blob_store: &BlobStore, named_blobs: &[(&[u8], Vec<u8>)]) {
|
||||
for (key, data) in named_blobs {
|
||||
assert_eq!(
|
||||
blob_store
|
||||
.get_blob(key, 0..usize::MAX)
|
||||
.await
|
||||
.unwrap()
|
||||
.as_ref(),
|
||||
Some(data),
|
||||
"Blob {} was not restored",
|
||||
String::from_utf8_lossy(key)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
struct Snapshot {
|
||||
keys: AHashSet<KeyValue>,
|
||||
@@ -240,7 +471,19 @@ struct KeyValue {
|
||||
|
||||
impl Snapshot {
|
||||
async fn new(db: &Store) -> Self {
|
||||
let is_sql = db.is_sql();
|
||||
Self::build(db, !db.is_sql(), true).await
|
||||
}
|
||||
|
||||
/// Comparable across backends: no counter values, which the SQL and
|
||||
/// key-value stores encode differently, and no blobs, which only live in
|
||||
/// the data store when it doubles as the blob store.
|
||||
#[cfg(all(feature = "rocks", feature = "sqlite"))]
|
||||
async fn new_portable(db: &Store) -> Self {
|
||||
Self::build(db, false, false).await
|
||||
}
|
||||
|
||||
async fn build(db: &Store, counter_values: bool, with_blobs: bool) -> Self {
|
||||
let is_sql = !counter_values;
|
||||
|
||||
let mut keys = AHashSet::new();
|
||||
|
||||
@@ -265,7 +508,12 @@ impl Snapshot {
|
||||
(SUBSPACE_QUOTA, !is_sql),
|
||||
(SUBSPACE_REPORT_OUT, true),
|
||||
(SUBSPACE_REPORT_IN, true),
|
||||
(SUBSPACE_DIRECTORY, true),
|
||||
(SUBSPACE_INBUXA, true),
|
||||
] {
|
||||
if subspace == SUBSPACE_BLOBS && !with_blobs {
|
||||
continue;
|
||||
}
|
||||
let from_key = AnyKey {
|
||||
subspace,
|
||||
key: vec![0u8],
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::{store::deflate_test_resource, utils::server::TestServer};
|
||||
@@ -122,6 +124,10 @@ pub async fn test(test: &TestServer) {
|
||||
println!("Running global id filtering tests...");
|
||||
test_global(store.clone()).await;
|
||||
|
||||
// inbuxa: trace documents as the index task builds them
|
||||
println!("Running trace document tests...");
|
||||
test_trace_documents(store.clone()).await;
|
||||
|
||||
// Large document insert test
|
||||
println!("Running large document insert tests...");
|
||||
let mut large_text = String::with_capacity(20 * 1024 * 1024);
|
||||
@@ -809,3 +815,160 @@ async fn test_global(store: SearchStore) {
|
||||
AHashSet::from_iter([3, 4, 5])
|
||||
);
|
||||
}
|
||||
|
||||
// inbuxa: MON-16: documents built by the index task from stored traces go
|
||||
// into every search backend (the SQL backends type etyp and qid as BIGINT)
|
||||
// and are found again by queue id and keyword.
|
||||
async fn test_trace_documents(store: SearchStore) {
|
||||
use registry::schema::{
|
||||
enums::SearchTracingField,
|
||||
structs::{
|
||||
Trace, TraceEvent, TraceKeyValue, TraceValue, TraceValueString,
|
||||
TraceValueUnsignedInt,
|
||||
},
|
||||
};
|
||||
use services::task_manager::index::trace_search_document;
|
||||
use trc::{DeliveryEvent, EventType, Key, SmtpEvent};
|
||||
|
||||
let kv_u = |key: Key, value: u64| TraceKeyValue {
|
||||
key,
|
||||
value: TraceValue::UnsignedInt(TraceValueUnsignedInt { value }),
|
||||
};
|
||||
let kv_s = |key: Key, value: &str| TraceKeyValue {
|
||||
key,
|
||||
value: TraceValue::String(TraceValueString {
|
||||
value: value.to_string(),
|
||||
}),
|
||||
};
|
||||
let event = |event: EventType, key_values: Vec<TraceKeyValue>| TraceEvent {
|
||||
event,
|
||||
key_values: key_values.into(),
|
||||
..Default::default()
|
||||
};
|
||||
let fields = [
|
||||
SearchTracingField::EventType,
|
||||
SearchTracingField::QueueId,
|
||||
SearchTracingField::Keywords,
|
||||
];
|
||||
|
||||
// An SMTP session that queued two messages, and a delivery attempt
|
||||
let session = Trace {
|
||||
events: vec![
|
||||
event(
|
||||
EventType::Smtp(SmtpEvent::ConnectionStart),
|
||||
vec![kv_s(Key::RemoteIp, "192.0.2.7")],
|
||||
),
|
||||
event(
|
||||
EventType::Smtp(SmtpEvent::MailFrom),
|
||||
vec![kv_s(Key::From, "[email protected]")],
|
||||
),
|
||||
event(
|
||||
EventType::Smtp(SmtpEvent::RcptTo),
|
||||
vec![kv_u(Key::QueueId, 9_000_000_001), kv_s(Key::To, "[email protected]")],
|
||||
),
|
||||
event(
|
||||
EventType::Smtp(SmtpEvent::RcptTo),
|
||||
vec![kv_u(Key::QueueId, 9_000_000_002)],
|
||||
),
|
||||
]
|
||||
.into(),
|
||||
};
|
||||
let delivery = Trace {
|
||||
events: vec![event(
|
||||
EventType::Delivery(DeliveryEvent::AttemptStart),
|
||||
vec![kv_u(Key::QueueId, 9_000_000_003), kv_s(Key::Hostname, "relay.example.net")],
|
||||
)]
|
||||
.into(),
|
||||
};
|
||||
let documents = vec![
|
||||
trace_search_document(100, &session, &fields),
|
||||
trace_search_document(101, &delivery, &fields),
|
||||
];
|
||||
assert!(
|
||||
documents
|
||||
.iter()
|
||||
.all(|d| d.has_field(&SearchField::Tracing(TracingSearchField::QueueId))
|
||||
&& d.has_field(&SearchField::Tracing(TracingSearchField::EventType))),
|
||||
"trace documents carry a queue id and an event type"
|
||||
);
|
||||
store.index(documents).await.unwrap();
|
||||
if let SearchStore::ElasticSearch(store) = &store {
|
||||
store.refresh_index(SearchIndex::Tracing).await.unwrap();
|
||||
}
|
||||
|
||||
let query = |filters: Vec<SearchFilter>| {
|
||||
let store = store.clone();
|
||||
async move {
|
||||
store
|
||||
.query_global(
|
||||
SearchQuery::new(SearchIndex::Tracing)
|
||||
.with_filter(SearchFilter::ge(SearchField::Id, 100u64))
|
||||
.with_filters(filters),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.collect::<AHashSet<_>>()
|
||||
}
|
||||
};
|
||||
// By queue id, the way x:Trace/query asks: the queue id column, or any
|
||||
// queue id in the keywords
|
||||
let by_queue_id = |queue_id: u64| {
|
||||
vec![
|
||||
SearchFilter::Or,
|
||||
SearchFilter::eq(TracingSearchField::QueueId, queue_id),
|
||||
SearchFilter::has_text(
|
||||
TracingSearchField::Keywords,
|
||||
queue_id.to_string(),
|
||||
Language::None,
|
||||
),
|
||||
SearchFilter::End,
|
||||
]
|
||||
};
|
||||
assert_eq!(query(by_queue_id(9_000_000_001)).await, AHashSet::from_iter([100]));
|
||||
assert_eq!(query(by_queue_id(9_000_000_002)).await, AHashSet::from_iter([100]));
|
||||
assert_eq!(query(by_queue_id(9_000_000_003)).await, AHashSet::from_iter([101]));
|
||||
assert_eq!(query(by_queue_id(9_000_000_004)).await, AHashSet::new());
|
||||
assert_eq!(
|
||||
query(vec![SearchFilter::eq(TracingSearchField::QueueId, 9_000_000_003u64)]).await,
|
||||
AHashSet::from_iter([101])
|
||||
);
|
||||
// By opening event type
|
||||
assert_eq!(
|
||||
query(vec![SearchFilter::eq(
|
||||
TracingSearchField::EventType,
|
||||
EventType::Delivery(DeliveryEvent::AttemptStart).to_id() as u64,
|
||||
)])
|
||||
.await,
|
||||
AHashSet::from_iter([101])
|
||||
);
|
||||
// By keyword: an address, lowercased, and its domain
|
||||
assert_eq!(
|
||||
query(vec![SearchFilter::has_text(
|
||||
TracingSearchField::Keywords,
|
||||
"example.org",
|
||||
Language::None,
|
||||
)])
|
||||
.await,
|
||||
AHashSet::from_iter([100])
|
||||
);
|
||||
assert_eq!(
|
||||
query(vec![SearchFilter::has_text(
|
||||
TracingSearchField::Keywords,
|
||||
"relay.example.net",
|
||||
Language::None,
|
||||
)])
|
||||
.await,
|
||||
AHashSet::from_iter([101])
|
||||
);
|
||||
|
||||
for id in [100u64, 101] {
|
||||
store
|
||||
.unindex(
|
||||
SearchQuery::new(SearchIndex::Tracing)
|
||||
.with_filter(SearchFilter::eq(SearchField::Id, id)),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -148,6 +148,73 @@ pub async fn test(test: &mut TestServer) {
|
||||
"test 9: to"
|
||||
);
|
||||
|
||||
// MON-16: the queueId filter finds the traces that name a queue id (the
|
||||
// session that queued the message and its delivery attempt) through the
|
||||
// search index, given as a string or a number (the index column is an
|
||||
// integer)
|
||||
fn queue_ids(value: &Value, out: &mut Vec<u64>) {
|
||||
match value {
|
||||
Value::Object(map) => {
|
||||
if map.get("key").and_then(|k| k.as_str()) == Some("queueId")
|
||||
&& let Some(id) = map
|
||||
.get("value")
|
||||
.and_then(|v| v.get("value").unwrap_or(v).as_u64())
|
||||
{
|
||||
out.push(id);
|
||||
}
|
||||
map.values().for_each(|v| queue_ids(v, out));
|
||||
}
|
||||
Value::Array(list) => list.iter().for_each(|v| queue_ids(v, out)),
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
let with_ids = traces
|
||||
.iter()
|
||||
.map(|t| {
|
||||
let mut ids = Vec::new();
|
||||
queue_ids(t, &mut ids);
|
||||
(t["id"].as_str().unwrap().to_string(), ids)
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let queue_id = with_ids
|
||||
.iter()
|
||||
.find_map(|(_, ids)| ids.first().copied())
|
||||
.expect("MON-16: a trace with a queue id");
|
||||
let mut expected = with_ids
|
||||
.iter()
|
||||
.filter(|(_, ids)| ids.contains(&queue_id))
|
||||
.map(|(id, _)| id.clone())
|
||||
.collect::<Vec<_>>();
|
||||
expected.sort();
|
||||
for filter in [json!(queue_id.to_string()), json!(queue_id)] {
|
||||
let response = admin
|
||||
.jmap_method_call("x:Trace/query", json!({"filter": {"queueId": filter}}))
|
||||
.await;
|
||||
let mut found = response
|
||||
.0
|
||||
.pointer("/methodResponses/0/1/ids")
|
||||
.and_then(|ids| ids.as_array())
|
||||
.map(|ids| {
|
||||
ids.iter()
|
||||
.filter_map(|id| id.as_str().map(str::to_string))
|
||||
.collect::<Vec<_>>()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
found.sort();
|
||||
assert_eq!(found, expected, "MON-16: queueId {filter}: {response:?}");
|
||||
}
|
||||
let response = admin
|
||||
.jmap_method_call(
|
||||
"x:Trace/query",
|
||||
json!({"filter": {"queueId": (queue_id ^ 0x5a5a_5a5a).to_string()}}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
response.0.pointer("/methodResponses/0/1/ids"),
|
||||
Some(&json!([])),
|
||||
"MON-16: an unknown queue id"
|
||||
);
|
||||
|
||||
// Acceptance test 24: destroy removes a trace; create is refused
|
||||
let trace_id = traces[0]["id"].as_str().unwrap().to_string();
|
||||
let response = admin
|
||||
|
||||
@@ -14,6 +14,12 @@ simply there -- and the release build failed on
|
||||
|
||||
after a tag had already been pushed. This is seconds, and it runs beside the
|
||||
other fork checks rather than waiting for a release to find out.
|
||||
|
||||
Being in the context isn't enough on its own: the Dockerfile cooks the
|
||||
dependencies (`cargo chef cook`) before it copies the tree in, from a recipe
|
||||
that carries only the workspace's manifests. So each patched path must also be
|
||||
copied into that stage before the cook step, or the same error comes back
|
||||
there -- as it did for 2026.9.24.2, the first tag after the context fix.
|
||||
"""
|
||||
|
||||
import re
|
||||
@@ -49,6 +55,26 @@ def allowed(dockerignore: Path) -> set[str]:
|
||||
return keep
|
||||
|
||||
|
||||
def copied_before_cook(dockerfile: Path) -> list[str] | None:
|
||||
"""Sources COPY'd into the stage that runs `cargo chef cook`, before it.
|
||||
|
||||
None when no stage cooks. A `COPY . .` covers everything.
|
||||
"""
|
||||
stage: list[str] = []
|
||||
for line in dockerfile.read_text().splitlines():
|
||||
stripped = line.strip()
|
||||
if re.match(r"(?i)^FROM\s", stripped):
|
||||
stage = []
|
||||
continue
|
||||
if "cargo chef cook" in stripped:
|
||||
return stage
|
||||
m = re.match(r"(?i)^COPY\s+(?!--from)(.+)$", stripped)
|
||||
if m:
|
||||
parts = m.group(1).split()
|
||||
stage.extend(p.strip("./").split("/")[0] or "." for p in parts[:-1])
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
paths = patched_paths(root / "Cargo.toml")
|
||||
if not paths:
|
||||
@@ -72,9 +98,21 @@ def main() -> int:
|
||||
f" Add `!{top}` to .dockerignore.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
copied = copied_before_cook(root / "Dockerfile")
|
||||
if copied is not None and "." not in copied:
|
||||
for p in paths:
|
||||
top = p.strip("/").split("/")[0]
|
||||
if top not in copied:
|
||||
print(
|
||||
f"Cargo.toml patches {p}, but the Dockerfile doesn't copy {top!r} into the\n"
|
||||
f" stage that runs `cargo chef cook` before that step, so cooking the\n"
|
||||
f" dependencies fails on it. Add `COPY {top}/ {top}/` before the cook.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
bad.append((p, top))
|
||||
if bad:
|
||||
return 1
|
||||
print(f"build context includes every patched path: {', '.join(paths)}")
|
||||
print(f"build context and cook stage include every patched path: {', '.join(paths)}")
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user