mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-29 23:37:04 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 7ed460ed21 | |||
| bcd017aa97 | |||
| e371b8aaa6 |
@@ -1,224 +1,59 @@
|
||||
# Copyright 2024 RustFS Team
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
name: Windows Filesystem Tests
|
||||
|
||||
name: Heal ownership interaction red-id
|
||||
on:
|
||||
push:
|
||||
branches: [ main ]
|
||||
paths:
|
||||
- "crates/ecstore/src/disk/**"
|
||||
- "crates/ecstore/src/layout/**"
|
||||
- "crates/ecstore/src/store/init_format.rs"
|
||||
- "crates/ecstore/src/set_disk/ops/locking.rs"
|
||||
- "crates/ecstore/src/object_api/write_commit_context.rs"
|
||||
- "crates/ecstore/src/bucket/utils.rs"
|
||||
- "crates/ecstore/Cargo.toml"
|
||||
- "crates/utils/src/dunce.rs"
|
||||
- "crates/utils/src/os/**"
|
||||
- "crates/utils/Cargo.toml"
|
||||
- "crates/utils/src/path.rs"
|
||||
- "rustfs/build.rs"
|
||||
- "rustfs/Cargo.toml"
|
||||
- "rustfs/src/main.rs"
|
||||
- "rustfs/src/server/runtime.rs"
|
||||
- "rustfs/src/startup_*.rs"
|
||||
- "Cargo.toml"
|
||||
- "Cargo.lock"
|
||||
- ".github/actions/setup/**"
|
||||
- ".github/workflows/windows-filesystem.yml"
|
||||
pull_request:
|
||||
branches: [ main ]
|
||||
paths:
|
||||
- "crates/ecstore/src/disk/**"
|
||||
- "crates/ecstore/src/layout/**"
|
||||
- "crates/ecstore/src/store/init_format.rs"
|
||||
- "crates/ecstore/src/set_disk/ops/locking.rs"
|
||||
- "crates/ecstore/src/object_api/write_commit_context.rs"
|
||||
- "crates/ecstore/src/bucket/utils.rs"
|
||||
- "crates/ecstore/Cargo.toml"
|
||||
- "crates/utils/src/dunce.rs"
|
||||
- "crates/utils/src/os/**"
|
||||
- "crates/utils/Cargo.toml"
|
||||
- "crates/utils/src/path.rs"
|
||||
- "rustfs/build.rs"
|
||||
- "rustfs/Cargo.toml"
|
||||
- "rustfs/src/main.rs"
|
||||
- "rustfs/src/server/runtime.rs"
|
||||
- "rustfs/src/startup_*.rs"
|
||||
- "Cargo.toml"
|
||||
- "Cargo.lock"
|
||||
- ".github/actions/setup/**"
|
||||
- ".github/workflows/windows-filesystem.yml"
|
||||
workflow_dispatch:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
|
||||
cancel-in-progress: true
|
||||
|
||||
env:
|
||||
CARGO_TERM_COLOR: always
|
||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
||||
RUST_BACKTRACE: 1
|
||||
|
||||
jobs:
|
||||
rename-safety:
|
||||
name: Rename Safety
|
||||
runs-on: windows-latest
|
||||
timeout-minutes: 60
|
||||
focused:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 120
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
|
||||
- name: Checkout
|
||||
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
|
||||
with:
|
||||
fetch-depth: 3
|
||||
persist-credentials: false
|
||||
|
||||
- name: Setup Rust environment
|
||||
- name: Verify source snapshot
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
actual_head=$(git rev-parse HEAD)
|
||||
actual_source=$(git rev-parse HEAD^)
|
||||
actual_base=$(git rev-parse HEAD^^)
|
||||
actual_tree=$(git rev-parse 'HEAD^^{tree}')
|
||||
actual_diff=$(git diff --full-index --binary "$actual_base" "$actual_source" | sha256sum | cut -d' ' -f1)
|
||||
actual_files=$(git diff --name-only "$actual_source" "$actual_head")
|
||||
printf 'head %s source %s base %s tree %s diff %s files %s\n' "$actual_head" "$actual_source" "$actual_base" "$actual_tree" "$actual_diff" "$actual_files"
|
||||
[[ "$actual_head" == "$GITHUB_SHA" ]]
|
||||
[[ "$actual_source" == "bcd017aa9763642d49a9fab61757431ecf3fd3ac" ]]
|
||||
[[ "$actual_base" == "e371b8aaa6d0e1436795c1415eaaba037717f99d" ]]
|
||||
[[ "$actual_tree" == "3b73e4f8ff15b8980a32a54c9145079162a06390" ]]
|
||||
[[ "$actual_diff" == "dad65272388e62e8a6a91e5a54aef2751c212ccb74282bdcdc61164c2344176c" ]]
|
||||
[[ "$actual_files" == ".github/workflows/windows-filesystem.yml" ]]
|
||||
- name: Setup Rust
|
||||
uses: ./.github/actions/setup
|
||||
with:
|
||||
rust-version: stable
|
||||
cache-shared-key: build-x86_64-pc-windows-msvc
|
||||
cache-shared-key: ci-e2e-distributed-hosted
|
||||
cache-save-if: 'false'
|
||||
install-build-packaging-tools: 'false'
|
||||
install-test-tools: 'false'
|
||||
|
||||
- name: Check production Windows dependencies
|
||||
shell: pwsh
|
||||
run: cargo check -p rustfs-ecstore --lib
|
||||
|
||||
- name: Test guarded rename publication
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib rename_all_ -- --nocapture
|
||||
|
||||
- name: Test multipart parent durability barriers
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
|
||||
|
||||
- name: Test Windows path conversion
|
||||
shell: pwsh
|
||||
run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
|
||||
|
||||
- name: Test Windows volume and disk-space probes
|
||||
shell: pwsh
|
||||
- name: Run the selected test
|
||||
shell: bash
|
||||
run: |
|
||||
# Map a drive letter to the temp drive's administrative share so the
|
||||
# remote-volume probe also covers mapped network drives.
|
||||
$drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
|
||||
net use R: "\\localhost\$drive`$" /persistent:no
|
||||
if ($LASTEXITCODE -eq 0) {
|
||||
$env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
|
||||
} else {
|
||||
Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
|
||||
}
|
||||
cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
|
||||
$status = $LASTEXITCODE
|
||||
if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
|
||||
exit $status
|
||||
|
||||
- name: Test Windows handle guards
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib windows_ -- --nocapture
|
||||
|
||||
- name: Test endpoint path parsing
|
||||
shell: pwsh
|
||||
run: "cargo test -p rustfs-ecstore --lib layout::endpoint:: -- --nocapture"
|
||||
|
||||
- name: Test startup temporary-directory cleanup
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib cleanup_tmp_on_startup_ -- --nocapture
|
||||
|
||||
- name: Test object key Windows admission
|
||||
shell: pwsh
|
||||
run: "cargo test -p rustfs-ecstore --lib bucket::utils:: -- --nocapture"
|
||||
|
||||
- name: Test Windows file-name rules
|
||||
shell: pwsh
|
||||
run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
|
||||
|
||||
- name: Test fresh format publication
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib fresh_format_load_initializes_all_disks -- --nocapture
|
||||
|
||||
# The native Windows main thread only gets the 1 MiB PE stack reserve, while
|
||||
# Linux and macOS hand out 8 MiB. Startup runs the whole root future on the
|
||||
# thread that calls `block_on`, so a debug server once overflowed before it
|
||||
# listened on Windows only. Start a real server to keep that covered.
|
||||
startup-smoke:
|
||||
name: Startup Smoke
|
||||
runs-on: windows-latest
|
||||
timeout-minutes: 90
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
|
||||
with:
|
||||
persist-credentials: false
|
||||
|
||||
- name: Setup Rust environment
|
||||
uses: ./.github/actions/setup
|
||||
with:
|
||||
rust-version: stable
|
||||
cache-shared-key: build-x86_64-pc-windows-msvc
|
||||
cache-save-if: 'false'
|
||||
install-build-packaging-tools: 'false'
|
||||
install-test-tools: 'false'
|
||||
|
||||
- name: Build debug server binary
|
||||
shell: pwsh
|
||||
run: cargo build --locked -p rustfs --bin rustfs
|
||||
|
||||
- name: Print version
|
||||
shell: pwsh
|
||||
run: |
|
||||
& .\target\debug\rustfs.exe --version
|
||||
if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
|
||||
|
||||
- name: Start server and wait for readiness
|
||||
shell: pwsh
|
||||
env:
|
||||
RUSTFS_ACCESS_KEY: rustfsadmin
|
||||
RUSTFS_SECRET_KEY: rustfsadmin
|
||||
run: |
|
||||
$volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
|
||||
$stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
|
||||
$stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
|
||||
New-Item -ItemType Directory -Force -Path $volume | Out-Null
|
||||
$server = Start-Process -FilePath .\target\debug\rustfs.exe `
|
||||
-ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
|
||||
-RedirectStandardOutput $stdout -RedirectStandardError $stderr `
|
||||
-NoNewWindow -PassThru
|
||||
# Cache the process handle now; otherwise ExitCode reads back empty.
|
||||
$null = $server.Handle
|
||||
$ready = $false
|
||||
try {
|
||||
for ($i = 0; $i -lt 180; $i++) {
|
||||
if ($server.HasExited) { break }
|
||||
try {
|
||||
$response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
|
||||
if ($response.StatusCode -eq 200) { $ready = $true; break }
|
||||
} catch {}
|
||||
Start-Sleep -Seconds 1
|
||||
}
|
||||
} finally {
|
||||
if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
|
||||
$server.WaitForExit()
|
||||
Write-Host '--- server stdout (tail) ---'
|
||||
Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
|
||||
Write-Host '--- server stderr (tail) ---'
|
||||
Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
|
||||
}
|
||||
if (-not $ready) {
|
||||
throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
|
||||
}
|
||||
set -euo pipefail
|
||||
export CARGO_TERM_COLOR=never CARGO_BUILD_JOBS=2 CARGO_PROFILE_TEST_DEBUG=0 CARGO_INCREMENTAL=0 RUST_BACKTRACE=0
|
||||
name="heal::manager::tests::admin_overlap::forced_id_conflict_preserves_scanner_queued_owner"
|
||||
log="$RUNNER_TEMP/interaction.log"
|
||||
set +e
|
||||
cargo test --locked -p rustfs-heal --lib "$name" -- --exact --test-threads=1 >"$log" 2>&1
|
||||
status=$?
|
||||
set -e
|
||||
tail -n 160 "$log"
|
||||
[[ "$status" -eq 101 ]]
|
||||
grep -Fxq 'running 1 test' "$log"
|
||||
grep -Fxq "test ${name} ... FAILED" "$log"
|
||||
grep -Fq 'left: Accepted' "$log"
|
||||
grep -Fq 'right: Dropped(AlreadyRunning)' "$log"
|
||||
grep -Fq 'test result: FAILED. 0 passed; 1 failed;' "$log"
|
||||
printf 'Confirmed expected missing ID recheck failure\n'
|
||||
|
||||
@@ -2282,13 +2282,22 @@ impl HealManager {
|
||||
None
|
||||
};
|
||||
|
||||
let scanner_queued_owner_required = request.source == HealRequestSource::Scanner
|
||||
&& mrf_notice_target.is_none()
|
||||
&& request.expected_mrf_bucket_incarnation_id.is_none();
|
||||
let duplicate = (!request.force_start && rejection.is_none()).then(|| {
|
||||
active_heal_for_dedup_key(&active_heals, &dedup_key)
|
||||
.map(|(task_id, _)| (task_id, "active"))
|
||||
.or_else(|| {
|
||||
queue
|
||||
.queued_request_id_for_dedup_key(&dedup_key)
|
||||
.map(|queued_id| (queued_id.to_string(), "queued"))
|
||||
let queued_id = if scanner_queued_owner_required {
|
||||
queue.scanner_queued_request_id_for_dedup_key(&dedup_key)
|
||||
} else {
|
||||
queue.queued_request_id_for_dedup_key(&dedup_key)
|
||||
};
|
||||
queued_id.map(|queued_id| (queued_id.to_string(), "queued")).or_else(|| {
|
||||
(scanner_queued_owner_required && queue.dedup_keys.contains_key(&dedup_key))
|
||||
.then(|| (request.id.clone(), "queued_owner_unknown"))
|
||||
})
|
||||
})
|
||||
.or_else(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| (task_id, "retrying")))
|
||||
});
|
||||
@@ -2296,12 +2305,12 @@ impl HealManager {
|
||||
// HS-06: under the minio_error overlap policy an exact duplicate
|
||||
// admin start reports the typed AlreadyRunning rejection instead
|
||||
// of the silent merge (MinIO's ErrHealAlreadyRunning).
|
||||
let mut admission = if (request.source == HealRequestSource::Admin
|
||||
&& config.overlap_policy == HealOverlapPolicy::MinioError)
|
||||
let mut admission = if duplicate_state == "queued_owner_unknown"
|
||||
|| (request.source == HealRequestSource::Admin && config.overlap_policy == HealOverlapPolicy::MinioError)
|
||||
|| (duplicate_state != "queued" && mrf_notice_target.as_ref().is_some_and(|target| target.kind.is_durable()))
|
||||
{
|
||||
// A running/retrying task may have observed the object before
|
||||
// this write committed. Its receipt cannot prove the new lease.
|
||||
// An unknown queued owner cannot certify a receipt. A running or
|
||||
// retrying task may have observed the object before this write.
|
||||
HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
|
||||
} else {
|
||||
Self::duplicate_admission_for_request(&request, &config)
|
||||
@@ -2514,10 +2523,17 @@ impl HealManager {
|
||||
.remove(&request_id, &request_heal_type, request_source)
|
||||
.await?;
|
||||
}
|
||||
if admission == HealAdmissionResult::Merged
|
||||
&& let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key)
|
||||
{
|
||||
task_id = queued_id.to_owned();
|
||||
if admission == HealAdmissionResult::Merged {
|
||||
let queued_id = if scanner_queued_owner_required {
|
||||
queue.scanner_queued_request_id_for_dedup_key(&dedup_key)
|
||||
} else {
|
||||
queue.queued_request_id_for_dedup_key(&dedup_key)
|
||||
};
|
||||
if let Some(queued_id) = queued_id {
|
||||
task_id = queued_id.to_owned();
|
||||
} else if scanner_queued_owner_required {
|
||||
return Err(Error::Other("Scanner queued owner is unavailable".to_owned()));
|
||||
}
|
||||
}
|
||||
let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
|
||||
let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
|
||||
|
||||
@@ -13,15 +13,23 @@
|
||||
// limitations under the License.
|
||||
/// The priority heal queue and its per-key dedup index.
|
||||
use super::*;
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
/// Per-key bookkeeping for the queued-request dedup index: how many queued
|
||||
/// requests hold the key, and the id of the first request that opened it —
|
||||
/// the O(1) stand-in for the former heap scan when a merge receipt needs to
|
||||
/// name a queued representative.
|
||||
/// Per-key bookkeeping for the queued-request dedup index. The legacy
|
||||
/// representative is retained for existing non-scanner callers; scanner
|
||||
/// receipts use the separately tracked queued holders.
|
||||
#[derive(Debug)]
|
||||
pub(super) struct DedupKeyEntry {
|
||||
pub(super) refcount: usize,
|
||||
pub(super) representative_request_id: String,
|
||||
pub(super) scanner_queued_owners: ScannerQueuedOwners,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(super) enum ScannerQueuedOwners {
|
||||
Single(u64),
|
||||
// Once a key has forced duplicates, keep every live holder until it is released.
|
||||
Indexed(BTreeMap<u64, String>),
|
||||
}
|
||||
|
||||
/// Priority queue wrapper for heal requests
|
||||
@@ -306,11 +314,7 @@ impl PriorityHealQueue {
|
||||
|
||||
#[cfg(test)]
|
||||
pub(super) fn pop_next(&mut self) -> Option<HealRequest> {
|
||||
self.heap.pop().map(|item| {
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
|
||||
self.remove_best_effort_count(item.request.source);
|
||||
item.request
|
||||
})
|
||||
self.heap.pop().map(|item| self.remove_queued_item(item))
|
||||
}
|
||||
|
||||
pub(super) fn is_empty(&self) -> bool {
|
||||
@@ -332,14 +336,32 @@ impl PriorityHealQueue {
|
||||
// also reserves the dedup key for later non-forced duplicates. The first
|
||||
// request that opens the key becomes the named representative for merge
|
||||
// receipts (taken before `request` moves into the heap).
|
||||
self.dedup_keys
|
||||
.entry(key.clone())
|
||||
.or_insert_with(|| DedupKeyEntry {
|
||||
refcount: 0,
|
||||
representative_request_id: request.id.clone(),
|
||||
})
|
||||
.refcount += 1;
|
||||
self.sequence += 1;
|
||||
if let Some(entry) = self.dedup_keys.get_mut(&key) {
|
||||
match &mut entry.scanner_queued_owners {
|
||||
ScannerQueuedOwners::Single(first_sequence) => {
|
||||
let mut owners = BTreeMap::new();
|
||||
if entry.refcount == 1 {
|
||||
owners.insert(*first_sequence, entry.representative_request_id.clone());
|
||||
}
|
||||
owners.insert(self.sequence, request.id.clone());
|
||||
entry.scanner_queued_owners = ScannerQueuedOwners::Indexed(owners);
|
||||
}
|
||||
ScannerQueuedOwners::Indexed(owners) => {
|
||||
owners.insert(self.sequence, request.id.clone());
|
||||
}
|
||||
}
|
||||
entry.refcount += 1;
|
||||
} else {
|
||||
self.dedup_keys.insert(
|
||||
key.clone(),
|
||||
DedupKeyEntry {
|
||||
refcount: 1,
|
||||
representative_request_id: request.id.clone(),
|
||||
scanner_queued_owners: ScannerQueuedOwners::Single(self.sequence),
|
||||
},
|
||||
);
|
||||
}
|
||||
let source = request.source;
|
||||
self.heap.push(PriorityQueueItem {
|
||||
priority: request.priority,
|
||||
@@ -402,7 +424,7 @@ impl PriorityHealQueue {
|
||||
self.heap = retained;
|
||||
|
||||
let displaced = displaced.map(|item| {
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
|
||||
self.remove_best_effort_count(item.request.source);
|
||||
self.refresh_dedup_representative(&item.dedup_key);
|
||||
item.request
|
||||
@@ -446,11 +468,7 @@ impl PriorityHealQueue {
|
||||
|
||||
#[cfg(test)]
|
||||
pub(super) fn pop(&mut self) -> Option<HealRequest> {
|
||||
self.heap.pop().map(|item| {
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
|
||||
self.remove_best_effort_count(item.request.source);
|
||||
item.request
|
||||
})
|
||||
self.heap.pop().map(|item| self.remove_queued_item(item))
|
||||
}
|
||||
|
||||
pub(super) fn pop_runnable_with_skips<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
|
||||
@@ -475,14 +493,7 @@ impl PriorityHealQueue {
|
||||
|
||||
self.restore_deferred_items(deferred);
|
||||
|
||||
(
|
||||
selected.map(|item| {
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
|
||||
self.remove_best_effort_count(item.request.source);
|
||||
item.request
|
||||
}),
|
||||
skipped,
|
||||
)
|
||||
(selected.map(|item| self.remove_queued_item(item)), skipped)
|
||||
}
|
||||
|
||||
pub(super) fn pop_runnable_with_fairness<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
|
||||
@@ -554,15 +565,19 @@ impl PriorityHealQueue {
|
||||
self.heap = BinaryHeap::from(items);
|
||||
|
||||
let chosen = chosen.map(|item| {
|
||||
let request = item.request;
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
|
||||
self.remove_best_effort_count(request.source);
|
||||
let request = self.remove_queued_item(item);
|
||||
self.record_dispatch(&request);
|
||||
request
|
||||
});
|
||||
(chosen, skipped)
|
||||
}
|
||||
|
||||
fn remove_queued_item(&mut self, item: PriorityQueueItem) -> HealRequest {
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
|
||||
self.remove_best_effort_count(item.request.source);
|
||||
item.request
|
||||
}
|
||||
|
||||
fn record_dispatch(&mut self, request: &HealRequest) {
|
||||
if is_best_effort_source(request.source) {
|
||||
self.mrf_normal_dispatches_since_best_effort = 0;
|
||||
@@ -667,10 +682,19 @@ impl PriorityHealQueue {
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, key: &str) {
|
||||
if let Some(entry) = dedup_keys.get_mut(key) {
|
||||
fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, item: &PriorityQueueItem) {
|
||||
if let Some(entry) = dedup_keys.get_mut(&item.dedup_key) {
|
||||
match &mut entry.scanner_queued_owners {
|
||||
ScannerQueuedOwners::Single(sequence) if entry.refcount != 1 || *sequence != item.sequence => {
|
||||
entry.scanner_queued_owners = ScannerQueuedOwners::Indexed(BTreeMap::new());
|
||||
}
|
||||
ScannerQueuedOwners::Single(_) => {}
|
||||
ScannerQueuedOwners::Indexed(owners) => {
|
||||
owners.remove(&item.sequence);
|
||||
}
|
||||
}
|
||||
if entry.refcount <= 1 {
|
||||
dedup_keys.remove(key);
|
||||
dedup_keys.remove(&item.dedup_key);
|
||||
} else {
|
||||
entry.refcount -= 1;
|
||||
}
|
||||
@@ -747,15 +771,19 @@ impl PriorityHealQueue {
|
||||
self.dedup_keys.get(key).map(|entry| entry.representative_request_id.as_str())
|
||||
}
|
||||
|
||||
/// Re-elect the representative for `key` from the queue entries holding
|
||||
/// it. Needed after a holder leaves the queue *without* becoming active
|
||||
/// (canceled by id, or displaced): the former opener may be the request
|
||||
/// that just left, and a merge receipt must never name an id that
|
||||
/// resolves nowhere. The scheduler pop path does not need this — the
|
||||
/// popped request surfaces in `active_heals` under the same id and the
|
||||
/// duplicate pre-check consults active heals before the queue. No-op for
|
||||
/// released keys; the survivor scan only runs when a key still has
|
||||
/// holders, which under forced duplicates is the rare admin path.
|
||||
pub(super) fn scanner_queued_request_id_for_dedup_key(&self, key: &str) -> Option<&str> {
|
||||
let entry = self.dedup_keys.get(key)?;
|
||||
match &entry.scanner_queued_owners {
|
||||
ScannerQueuedOwners::Single(_) if entry.refcount == 1 => Some(entry.representative_request_id.as_str()),
|
||||
ScannerQueuedOwners::Indexed(owners) if owners.len() == entry.refcount => {
|
||||
owners.first_key_value().map(|(_, id)| id.as_str())
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Re-elect the legacy representative from the heap after cancellation or
|
||||
/// displacement. Dequeue leaves it unchanged for existing callers.
|
||||
pub(super) fn refresh_dedup_representative(&mut self, key: &str) {
|
||||
if !self.dedup_keys.contains_key(key) {
|
||||
return;
|
||||
@@ -786,7 +814,7 @@ impl PriorityHealQueue {
|
||||
while let Some(item) = self.heap.pop() {
|
||||
if removed.is_none() && item.request.id == request_id {
|
||||
let key = item.dedup_key.clone();
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
|
||||
self.remove_best_effort_count(item.request.source);
|
||||
affected_key = Some(key);
|
||||
removed = Some(item.request);
|
||||
@@ -812,7 +840,7 @@ impl PriorityHealQueue {
|
||||
|
||||
while let Some(item) = self.heap.pop() {
|
||||
if should_remove(&item.request) {
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
|
||||
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
|
||||
self.remove_best_effort_count(item.request.source);
|
||||
affected_keys.push(item.dedup_key);
|
||||
removed.push(item.request);
|
||||
|
||||
@@ -34,6 +34,7 @@ mod auto_scan;
|
||||
mod completion;
|
||||
mod disposition;
|
||||
mod legacy_ambiguity_logging;
|
||||
mod queued_representative;
|
||||
mod running_mainline;
|
||||
|
||||
use super::super::{DiskOption, DiskStore, Endpoint, new_disk, storage_api::status::BucketInfo};
|
||||
|
||||
@@ -450,6 +450,176 @@ async fn admin_overlap_normal_start_cannot_enter_force_start_cancellation_window
|
||||
assert_eq!(manager.heal_queue.lock().await.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn admin_force_start_rechecks_request_id_after_cancelling_overlaps() {
|
||||
for source in [HealRequestSource::Scanner, HealRequestSource::Internal] {
|
||||
let manager = manager_with_policy(HealOverlapPolicy::Merge);
|
||||
let previous_id = install_owner(&manager, admin_prefix_request("bucket", "scope/child/"), OwnerState::Active).await;
|
||||
let aliases = manager.task_aliases.lock().await;
|
||||
let mut replacement = admin_prefix_request("bucket", "scope/");
|
||||
replacement.force_start = true;
|
||||
let request_id = replacement.id.clone();
|
||||
let mut forced = Box::pin(manager.submit_heal_request_with_receipt(replacement));
|
||||
|
||||
tokio::time::timeout(
|
||||
Duration::from_secs(5),
|
||||
std::future::poll_fn(|cx| {
|
||||
assert!(
|
||||
std::pin::pin!(tokio::task::unconstrained(forced.as_mut()))
|
||||
.poll(cx)
|
||||
.is_pending()
|
||||
);
|
||||
if manager.active_heals.try_lock().is_ok() {
|
||||
Poll::Ready(())
|
||||
} else {
|
||||
cx.waker().wake_by_ref();
|
||||
Poll::Pending
|
||||
}
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.expect("forceStart reaches cancellation with registry locks released");
|
||||
|
||||
let mut competing = HealRequest::object("other-bucket".to_string(), "object".to_string(), None);
|
||||
competing.id = request_id.clone();
|
||||
competing.source = source;
|
||||
competing.options.recreate_missing = false;
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(competing)
|
||||
.await
|
||||
.expect("non-admin admission during forceStart cancellation");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
|
||||
assert_eq!(receipt.task_id, request_id);
|
||||
|
||||
drop(aliases);
|
||||
let receipt = forced.await.expect("forceStart returns a conflict");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning));
|
||||
assert_eq!(receipt.task_id, request_id);
|
||||
let queue = manager.heal_queue.lock().await;
|
||||
let same_id: Vec<_> = queue.requests().filter(|request| request.id == request_id).collect();
|
||||
assert_eq!(same_id.len(), 1, "conflicting requests must not share an id");
|
||||
assert_eq!(same_id[0].source, source);
|
||||
drop(queue);
|
||||
assert!(
|
||||
matches!(manager.get_task_status(&previous_id).await, Ok(HealTaskStatus::Cancelled)),
|
||||
"overlapping admin cancellation precedes the conflict recheck"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn forced_id_conflict_preserves_scanner_queued_owner() {
|
||||
for competing_source in [HealRequestSource::Scanner, HealRequestSource::Internal] {
|
||||
let manager = manager_with_policy(HealOverlapPolicy::Merge);
|
||||
let previous_id = install_owner(&manager, admin_object_request("object"), OwnerState::Active).await;
|
||||
let mut queued_ids = Vec::new();
|
||||
for source in [
|
||||
HealRequestSource::Scanner,
|
||||
HealRequestSource::Internal,
|
||||
HealRequestSource::Scanner,
|
||||
] {
|
||||
let mut request = HealRequest::object("bucket".to_string(), "object".to_string(), None);
|
||||
request.source = source;
|
||||
request.force_start = true;
|
||||
let id = request.id.clone();
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(request)
|
||||
.await
|
||||
.expect("queue forced work");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
|
||||
assert_eq!(receipt.task_id, id);
|
||||
queued_ids.push(id);
|
||||
}
|
||||
|
||||
// The first holder leaves the queue, leaving two forced tasks queued.
|
||||
let selected = {
|
||||
let mut queue = manager.heal_queue.lock().await;
|
||||
let (selected, skipped) = queue.pop_runnable_with_skips(|_| true, |_| None);
|
||||
assert!(skipped.is_empty());
|
||||
let selected = selected.expect("first queued task should run");
|
||||
assert_eq!(selected.id, queued_ids[0]);
|
||||
assert_eq!(queue.len(), 2);
|
||||
selected
|
||||
};
|
||||
let completed_task = HealTask::from_request(selected, manager.storage.clone());
|
||||
let mut completed = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await;
|
||||
Arc::make_mut(completed.outcome.as_mut().expect("completed task outcome")).finish(None);
|
||||
manager
|
||||
.completed_heals
|
||||
.lock()
|
||||
.await
|
||||
.insert(queued_ids[0].clone(), Arc::new(completed));
|
||||
|
||||
let aliases = manager.task_aliases.lock().await;
|
||||
let mut replacement = admin_object_request("object");
|
||||
replacement.force_start = true;
|
||||
let competing_id = replacement.id.clone();
|
||||
let mut forced = Box::pin(manager.submit_heal_request_with_receipt(replacement));
|
||||
tokio::time::timeout(
|
||||
Duration::from_secs(5),
|
||||
std::future::poll_fn(|cx| {
|
||||
assert!(
|
||||
std::pin::pin!(tokio::task::unconstrained(forced.as_mut()))
|
||||
.poll(cx)
|
||||
.is_pending()
|
||||
);
|
||||
if manager.active_heals.try_lock().is_ok() {
|
||||
Poll::Ready(())
|
||||
} else {
|
||||
cx.waker().wake_by_ref();
|
||||
Poll::Pending
|
||||
}
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.expect("forceStart reaches cancellation with registry locks released");
|
||||
|
||||
let mut competing = HealRequest::object("bucket".to_string(), "object".to_string(), None);
|
||||
competing.id = competing_id.clone();
|
||||
competing.source = competing_source;
|
||||
competing.force_start = true;
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(competing)
|
||||
.await
|
||||
.expect("competing same-key forced work");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
|
||||
assert_eq!(receipt.task_id, competing_id);
|
||||
|
||||
drop(aliases);
|
||||
let receipt = forced.await.expect("forceStart returns an ID conflict");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning));
|
||||
assert_eq!(receipt.task_id, competing_id);
|
||||
assert!(!manager.active_heals.lock().await.contains_key(&competing_id));
|
||||
assert!(!manager.retrying_heals.lock().await.contains_key(&competing_id));
|
||||
assert!(matches!(manager.get_task_status(&previous_id).await, Ok(HealTaskStatus::Cancelled)));
|
||||
{
|
||||
let queue = manager.heal_queue.lock().await;
|
||||
let same_id: Vec<_> = queue.requests().filter(|request| request.id == competing_id).collect();
|
||||
assert_eq!(same_id.len(), 1, "the competing ID must have one live owner");
|
||||
assert_eq!(same_id[0].source, competing_source);
|
||||
assert_eq!(queue.len(), 3);
|
||||
}
|
||||
|
||||
let mut scanner = HealRequest::object("bucket".to_string(), "object".to_string(), None);
|
||||
scanner.source = HealRequestSource::Scanner;
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(scanner)
|
||||
.await
|
||||
.expect("scanner should merge into queued work");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Merged);
|
||||
assert_ne!(receipt.task_id, queued_ids[0], "the completed holder cannot own a queued receipt");
|
||||
assert!(
|
||||
manager
|
||||
.heal_queue
|
||||
.lock()
|
||||
.await
|
||||
.requests()
|
||||
.any(|request| request.id == receipt.task_id),
|
||||
"the canonical receipt must name a live queued task"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn admin_overlap_cancel_cannot_miss_a_queued_to_active_transition() {
|
||||
let manager = manager_with_policy(HealOverlapPolicy::Merge);
|
||||
|
||||
@@ -0,0 +1,398 @@
|
||||
// Copyright 2026 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::*;
|
||||
|
||||
fn object_request(source: HealRequestSource, priority: HealPriority, force_start: bool) -> HealRequest {
|
||||
let mut request = HealRequest::new(
|
||||
HealType::Object {
|
||||
bucket: "canonical-owner".to_owned(),
|
||||
object: "object".to_owned(),
|
||||
version_id: None,
|
||||
},
|
||||
HealOptions::default(),
|
||||
priority,
|
||||
);
|
||||
request.source = source;
|
||||
request.force_start = force_start;
|
||||
request
|
||||
}
|
||||
|
||||
async fn scanner_receipt_tracks_queued_owner(use_fairness: bool) {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(
|
||||
Arc::new(MockStorage),
|
||||
Some(HealConfig {
|
||||
queue_size: 16,
|
||||
low_priority_merge_enable: true,
|
||||
overlap_policy: HealOverlapPolicy::Merge,
|
||||
..HealConfig::default()
|
||||
}),
|
||||
);
|
||||
|
||||
if use_fairness {
|
||||
let mut queue = manager.heal_queue.lock().await;
|
||||
for index in 0..4 {
|
||||
let request = bucket_request(&format!("internal-{index}"), HealPriority::Normal, HealRequestSource::Internal);
|
||||
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
|
||||
}
|
||||
for _ in 0..4 {
|
||||
let (request, _) = queue.pop_runnable_with_fairness(|_| true, |_| None);
|
||||
assert_eq!(request.expect("internal work should run").source, HealRequestSource::Internal);
|
||||
}
|
||||
}
|
||||
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let first_id = first.id.clone();
|
||||
let key = PriorityHealQueue::make_dedup_key(&first);
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
|
||||
let second_id = second.id.clone();
|
||||
for request in [first, second] {
|
||||
assert_eq!(
|
||||
manager
|
||||
.submit_heal_request_with_receipt(request)
|
||||
.await
|
||||
.expect("admit task")
|
||||
.result,
|
||||
HealAdmissionResult::Accepted
|
||||
);
|
||||
}
|
||||
|
||||
let first = {
|
||||
let mut queue = manager.heal_queue.lock().await;
|
||||
let (selected, skipped) = if use_fairness {
|
||||
queue.pop_runnable_with_fairness(|_| true, |_| None)
|
||||
} else {
|
||||
queue.pop_runnable_with_skips(|_| true, |_| None)
|
||||
};
|
||||
assert!(skipped.is_empty());
|
||||
let selected = selected.expect("the first same-priority task should run");
|
||||
assert_eq!(selected.id, first_id);
|
||||
assert!(queue.contains_request_id(&second_id));
|
||||
assert_eq!(
|
||||
queue.queued_request_id_for_dedup_key(&key),
|
||||
Some(first_id.as_str()),
|
||||
"the existing representative for other sources must remain unchanged"
|
||||
);
|
||||
selected
|
||||
};
|
||||
manager
|
||||
.active_heals
|
||||
.lock()
|
||||
.await
|
||||
.insert(first_id.clone(), Arc::new(HealTask::from_request(first, Arc::new(MockStorage))));
|
||||
|
||||
let active = manager
|
||||
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false))
|
||||
.await
|
||||
.expect("scanner admission while the first task is active");
|
||||
assert_eq!(active.result, HealAdmissionResult::Merged);
|
||||
assert_eq!(active.task_id, first_id, "active ownership keeps precedence");
|
||||
|
||||
let completed_task = manager
|
||||
.active_heals
|
||||
.lock()
|
||||
.await
|
||||
.remove(&first_id)
|
||||
.expect("first task should leave active ownership");
|
||||
let mut retained = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await;
|
||||
Arc::make_mut(retained.outcome.as_mut().expect("retained task outcome")).finish(None);
|
||||
manager
|
||||
.completed_heals
|
||||
.lock()
|
||||
.await
|
||||
.insert(first_id.clone(), Arc::new(retained));
|
||||
let queued = manager
|
||||
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false))
|
||||
.await
|
||||
.expect("scanner admission after the first task leaves active state");
|
||||
assert_eq!(queued.result, HealAdmissionResult::Merged);
|
||||
assert_eq!(queued.task_id, second_id, "the receipt must name the remaining queued task");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_receipt_tracks_queued_owner_after_normal_dequeue() {
|
||||
scanner_receipt_tracks_queued_owner(false).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_receipt_tracks_queued_owner_after_fair_dequeue() {
|
||||
scanner_receipt_tracks_queued_owner(true).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_receipt_keeps_queued_owner_when_a_different_forced_task_runs() {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false);
|
||||
let first_id = first.id.clone();
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
|
||||
let second_id = second.id.clone();
|
||||
{
|
||||
let mut queue = manager.heal_queue.lock().await;
|
||||
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
|
||||
assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
|
||||
let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
|
||||
assert_eq!(selected.expect("higher-priority forced task should run").id, second_id);
|
||||
}
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
|
||||
.await
|
||||
.expect("scanner admission after a non-representative task leaves the queue");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Merged);
|
||||
assert_eq!(receipt.task_id, first_id);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_receipt_reselects_after_the_queued_owner_is_cancelled() {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let first_id = first.id.clone();
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
|
||||
let second_id = second.id.clone();
|
||||
let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
|
||||
let third_id = third.id.clone();
|
||||
{
|
||||
let mut queue = manager.heal_queue.lock().await;
|
||||
for request in [first, second, third] {
|
||||
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
|
||||
}
|
||||
let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
|
||||
assert_eq!(selected.expect("first task should run").id, first_id);
|
||||
}
|
||||
let first_receipt = manager
|
||||
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
|
||||
.await
|
||||
.expect("first scanner admission");
|
||||
assert_eq!(first_receipt.result, HealAdmissionResult::Merged);
|
||||
assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id);
|
||||
let remaining = if first_receipt.task_id == second_id {
|
||||
third_id
|
||||
} else {
|
||||
second_id
|
||||
};
|
||||
manager.heal_queue.lock().await.remove_request_id(&first_receipt.task_id);
|
||||
|
||||
let next_receipt = manager
|
||||
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
|
||||
.await
|
||||
.expect("scanner admission after the queued owner is removed");
|
||||
assert_eq!(next_receipt.result, HealAdmissionResult::Merged);
|
||||
assert_eq!(next_receipt.task_id, remaining);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_receipt_reselects_after_multiple_forced_dequeues() {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let first_id = first.id.clone();
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
|
||||
let second_id = second.id.clone();
|
||||
let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
|
||||
let third_id = third.id.clone();
|
||||
{
|
||||
let mut queue = manager.heal_queue.lock().await;
|
||||
for request in [first, second, third] {
|
||||
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
|
||||
}
|
||||
let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
|
||||
assert_eq!(selected.expect("first task should run").id, first_id);
|
||||
}
|
||||
let first_receipt = manager
|
||||
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
|
||||
.await
|
||||
.expect("first scanner admission");
|
||||
assert_eq!(first_receipt.result, HealAdmissionResult::Merged);
|
||||
assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id);
|
||||
let remaining = if first_receipt.task_id == second_id {
|
||||
third_id
|
||||
} else {
|
||||
second_id
|
||||
};
|
||||
{
|
||||
let mut queue = manager.heal_queue.lock().await;
|
||||
let (selected, _) = queue.pop_runnable_with_skips(|request| request.id == first_receipt.task_id, |_| None);
|
||||
assert_eq!(selected.expect("selected queued owner should run").id, first_receipt.task_id);
|
||||
}
|
||||
let next_receipt = manager
|
||||
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
|
||||
.await
|
||||
.expect("scanner admission after another queued owner starts");
|
||||
assert_eq!(next_receipt.result, HealAdmissionResult::Merged);
|
||||
assert_eq!(next_receipt.task_id, remaining);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_owner_index_survives_a_single_remaining_holder_and_new_forced_work() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let first_id = first.id.clone();
|
||||
let key = PriorityHealQueue::make_dedup_key(&first);
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
|
||||
let second_id = second.id.clone();
|
||||
let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
|
||||
let third_id = third.id.clone();
|
||||
|
||||
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
|
||||
assert!(matches!(&queue.dedup_keys[&key].scanner_queued_owners, ScannerQueuedOwners::Single(_)));
|
||||
assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
|
||||
assert_eq!(queue.pop().expect("first queued task").id, first_id);
|
||||
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(first_id.as_str()));
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
|
||||
|
||||
assert_eq!(queue.push(third), QueuePushOutcome::Accepted);
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
|
||||
assert_eq!(queue.pop().expect("second queued task").id, second_id);
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_owner_index_distinguishes_holders_with_the_same_request_id() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let key = PriorityHealQueue::make_dedup_key(&first);
|
||||
let shared_id = first.id.clone();
|
||||
let mut second = object_request(HealRequestSource::Admin, HealPriority::High, true);
|
||||
second.id.clone_from(&shared_id);
|
||||
let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
|
||||
let third_id = third.id.clone();
|
||||
for request in [first, second, third] {
|
||||
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
|
||||
}
|
||||
|
||||
assert_eq!(queue.pop().expect("first holder").id, shared_id);
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(shared_id.as_str()));
|
||||
assert_eq!(queue.remove_request_id(&shared_id).expect("second holder").id, shared_id);
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
|
||||
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_owner_index_tracks_bulk_removal_and_key_release() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let first_id = first.id.clone();
|
||||
let key = PriorityHealQueue::make_dedup_key(&first);
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::Normal, true);
|
||||
let second_id = second.id.clone();
|
||||
let third = object_request(HealRequestSource::Internal, HealPriority::Low, true);
|
||||
let third_id = third.id.clone();
|
||||
for request in [first, second, third] {
|
||||
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
|
||||
}
|
||||
|
||||
let removed = queue.remove_matching(|request| request.id == first_id || request.id == second_id);
|
||||
assert_eq!(removed.len(), 2);
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
|
||||
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
|
||||
assert_eq!(queue.remove_matching(|request| request.id == third_id).len(), 1);
|
||||
assert!(!queue.dedup_keys.contains_key(&key));
|
||||
|
||||
let next = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let next_id = next.id.clone();
|
||||
assert_eq!(queue.push(next), QueuePushOutcome::Accepted);
|
||||
assert!(matches!(&queue.dedup_keys[&key].scanner_queued_owners, ScannerQueuedOwners::Single(_)));
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(next_id.as_str()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_owner_index_tracks_displacement_without_changing_legacy_reselection() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false);
|
||||
let first_id = first.id.clone();
|
||||
let key = PriorityHealQueue::make_dedup_key(&first);
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::Normal, true);
|
||||
let second_id = second.id.clone();
|
||||
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
|
||||
assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
|
||||
|
||||
let incoming = bucket_request("displacement", HealPriority::Urgent, HealRequestSource::Internal);
|
||||
let displaced = queue
|
||||
.push_displacing_lower_priority(incoming)
|
||||
.expect("lower priority task should be displaced");
|
||||
assert_eq!(displaced.id, first_id);
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
|
||||
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_receipt_rejects_a_missing_or_empty_holder_index() {
|
||||
for missing in [true, false] {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let key = PriorityHealQueue::make_dedup_key(&first);
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
|
||||
{
|
||||
let mut queue = manager.heal_queue.lock().await;
|
||||
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
|
||||
assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
|
||||
let entry = queue.dedup_keys.get_mut(&key).expect("dedup key should exist");
|
||||
entry.scanner_queued_owners = if missing {
|
||||
ScannerQueuedOwners::Single(0)
|
||||
} else {
|
||||
ScannerQueuedOwners::Indexed(std::collections::BTreeMap::new())
|
||||
};
|
||||
}
|
||||
|
||||
for leave_one in [false, true] {
|
||||
if leave_one {
|
||||
manager.heal_queue.lock().await.pop().expect("one holder should leave");
|
||||
}
|
||||
let request = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let request_id = request.id.clone();
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(request)
|
||||
.await
|
||||
.expect("admission should fail closed");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning));
|
||||
assert_eq!(receipt.task_id, request_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_and_legacy_owners_keep_their_distinct_reselection_orders() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let first_id = first.id.clone();
|
||||
let key = PriorityHealQueue::make_dedup_key(&first);
|
||||
let second = object_request(HealRequestSource::Admin, HealPriority::Low, true);
|
||||
let second_id = second.id.clone();
|
||||
let third = object_request(HealRequestSource::Internal, HealPriority::Normal, true);
|
||||
let third_id = third.id.clone();
|
||||
for request in [first, second, third] {
|
||||
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
|
||||
}
|
||||
|
||||
assert_eq!(queue.remove_request_id(&first_id).expect("first holder should be removed").id, first_id);
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
|
||||
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_owner_index_tracks_a_series_of_forced_holders() {
|
||||
let mut queue = PriorityHealQueue::new();
|
||||
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
|
||||
let key = PriorityHealQueue::make_dedup_key(&first);
|
||||
let mut ids = vec![first.id.clone()];
|
||||
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
|
||||
for _ in 0..128 {
|
||||
let request = object_request(HealRequestSource::Internal, HealPriority::High, true);
|
||||
ids.push(request.id.clone());
|
||||
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
|
||||
}
|
||||
for id in &ids {
|
||||
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(id.as_str()));
|
||||
assert_eq!(&queue.pop().expect("indexed holder should remain queued").id, id);
|
||||
}
|
||||
assert!(!queue.dedup_keys.contains_key(&key));
|
||||
}
|
||||
Reference in New Issue
Block a user