1494 lines
53 KiB
Rust
1494 lines
53 KiB
Rust
//! Typed commands and a single-owner worker for application operations.
|
|
|
|
use std::collections::VecDeque;
|
|
use std::ffi::OsString;
|
|
use std::path::PathBuf;
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
|
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender, TrySendError};
|
|
use std::thread::{self, JoinHandle};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use crate::app::Application;
|
|
use crate::app::{
|
|
FirstIssueAttempt, FirstIssueDeliveryOutcome, FirstIssueRetry, RestartChoice, RestartReport,
|
|
RestorePreview,
|
|
};
|
|
use crate::backup::{ArchiveSummary, BackupReason, HistoricalApplyStatus};
|
|
use crate::config::SettingsFile;
|
|
use crate::delivery::files::{FileOutputError, PreparedDestination};
|
|
use crate::delivery::{QrCodeOutput, QrError};
|
|
use crate::domain::{ClientAddresses, PublicKey};
|
|
use crate::runtime::RuntimeObservation;
|
|
use crate::vault::source::SelectedFileImportReport;
|
|
use crate::vault::source::SourceImportReport;
|
|
|
|
use super::state::{OperationKind, PendingOperation, SessionView};
|
|
|
|
pub const EVENT_QUEUE_CAPACITY: usize = 16;
|
|
pub const COMMAND_QUEUE_CAPACITY: usize = 4;
|
|
pub const DEFAULT_REFRESH_INTERVAL: Duration = Duration::from_secs(2);
|
|
pub const DEFAULT_READ_TIMEOUT: Duration = Duration::from_millis(500);
|
|
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub enum UiCommand {
|
|
CreateClient {
|
|
alias: String,
|
|
manual_addresses: Option<ClientAddresses>,
|
|
created_utc: Option<String>,
|
|
},
|
|
RenameClient {
|
|
identity: PublicKey,
|
|
alias: String,
|
|
},
|
|
DeleteClient {
|
|
identity: PublicKey,
|
|
},
|
|
DiscardDraft,
|
|
Commit,
|
|
CreateBackup,
|
|
ListBackups,
|
|
ImportSelectedFile {
|
|
file_name: OsString,
|
|
},
|
|
PreviewRestart,
|
|
Restart(RestartChoice),
|
|
PrepareDelivery {
|
|
identity: PublicKey,
|
|
path: PathBuf,
|
|
include_qr: bool,
|
|
reissue: bool,
|
|
attempt: FirstIssueAttempt,
|
|
},
|
|
DeliverFirstIssue {
|
|
attempt: FirstIssueAttempt,
|
|
overwrite_confirmed: bool,
|
|
prepared_generation: u64,
|
|
},
|
|
PreviewRestore {
|
|
file_name: String,
|
|
},
|
|
CancelPreparedDelivery,
|
|
CancelRestorePreview,
|
|
ConfirmRestore {
|
|
preview_generation: u64,
|
|
},
|
|
}
|
|
|
|
/// Secret-bearing text is carried only by a non-Debug, non-Clone owner and
|
|
/// zeroized when the worker finishes or drops the queued request.
|
|
pub struct SecretText(zeroize::Zeroizing<String>);
|
|
|
|
impl SecretText {
|
|
pub fn new(value: String) -> Self {
|
|
Self(zeroize::Zeroizing::new(value))
|
|
}
|
|
|
|
pub fn expose(&self) -> &str {
|
|
self.0.as_str()
|
|
}
|
|
}
|
|
|
|
pub enum SensitiveCommand {
|
|
PrepareInitialSetup(SecretText),
|
|
ReissueClient {
|
|
identity: PublicKey,
|
|
password: SecretText,
|
|
overwrite_confirmed: bool,
|
|
prepared_generation: u64,
|
|
},
|
|
}
|
|
|
|
impl UiCommand {
|
|
pub const fn kind(&self) -> OperationKind {
|
|
match self {
|
|
Self::CreateClient { .. } => OperationKind::CreateClient,
|
|
Self::RenameClient { .. } => OperationKind::RenameClient,
|
|
Self::DeleteClient { .. } => OperationKind::DeleteClient,
|
|
Self::DiscardDraft => OperationKind::DiscardDraft,
|
|
Self::Commit => OperationKind::Commit,
|
|
Self::CreateBackup => OperationKind::CreateBackup,
|
|
Self::ListBackups => OperationKind::ListBackups,
|
|
Self::ImportSelectedFile { .. } => OperationKind::ImportSelectedFile,
|
|
Self::PreviewRestart => OperationKind::PreviewRestart,
|
|
Self::Restart(_) => OperationKind::Restart,
|
|
Self::PrepareDelivery { .. } | Self::DeliverFirstIssue { .. } => {
|
|
OperationKind::DeliverClient
|
|
}
|
|
Self::PreviewRestore { .. }
|
|
| Self::CancelRestorePreview
|
|
| Self::ConfirmRestore { .. } => OperationKind::Restore,
|
|
Self::CancelPreparedDelivery => OperationKind::DeliverClient,
|
|
}
|
|
}
|
|
|
|
const fn changes_view(&self) -> bool {
|
|
matches!(
|
|
self,
|
|
Self::CreateClient { .. }
|
|
| Self::RenameClient { .. }
|
|
| Self::DeleteClient { .. }
|
|
| Self::DiscardDraft
|
|
| Self::Commit
|
|
| Self::ImportSelectedFile { .. }
|
|
| Self::Restart(_)
|
|
| Self::ConfirmRestore { .. }
|
|
)
|
|
}
|
|
}
|
|
|
|
impl SensitiveCommand {
|
|
const fn kind(&self) -> OperationKind {
|
|
match self {
|
|
Self::PrepareInitialSetup(_) => OperationKind::InitialSetup,
|
|
Self::ReissueClient { .. } => OperationKind::DeliverClient,
|
|
}
|
|
}
|
|
const fn changes_view(&self) -> bool {
|
|
matches!(self, Self::PrepareInitialSetup(_))
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum CommandResult {
|
|
ClientCreated(PublicKey),
|
|
ClientRenamed,
|
|
ClientDeleted { record_removed: bool },
|
|
DraftDiscarded,
|
|
Committed(PathBuf),
|
|
BackupCreated(PathBuf),
|
|
Backups(Vec<ArchiveSummary>),
|
|
SelectedFileImported(SelectedFileImportReport),
|
|
InitialSetupPrepared(SourceImportReport),
|
|
RestartPreview(RestartPreviewSummary),
|
|
Restarted(RestartReport),
|
|
DeliveryDestination(DeliveryDestinationSummary),
|
|
FirstIssue(FirstIssueSummary),
|
|
Reissued(ReissueSummary),
|
|
RestorePreview(RestorePreviewSummary),
|
|
Restored(PathBuf),
|
|
Cancelled,
|
|
}
|
|
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct RestartPreviewSummary {
|
|
pub has_draft: bool,
|
|
pub interface: String,
|
|
pub systemd_unit: String,
|
|
}
|
|
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct DeliveryDestinationSummary {
|
|
pub identity: PublicKey,
|
|
pub path: PathBuf,
|
|
pub overwrite_required: bool,
|
|
pub include_qr: bool,
|
|
pub reissue: bool,
|
|
pub attempt: FirstIssueAttempt,
|
|
pub prepared_generation: u64,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct FirstIssueSummary {
|
|
pub identity: PublicKey,
|
|
pub status: FirstIssueStatus,
|
|
pub retry: Option<FirstIssueRetry>,
|
|
pub activation_warning: bool,
|
|
pub qr: QrStatus,
|
|
pub qr_output: Option<QrCodeOutput>,
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
pub enum FirstIssueStatus {
|
|
Delivered,
|
|
PublishedButBookkeepingUncertain,
|
|
Failed(FileOutputError),
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
pub enum QrStatus {
|
|
NotRequested,
|
|
Ready,
|
|
Failed(QrError),
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct ReissueSummary {
|
|
pub activation_warning: bool,
|
|
pub qr: QrStatus,
|
|
pub qr_output: Option<QrCodeOutput>,
|
|
}
|
|
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct RestorePreviewSummary {
|
|
pub file_name: String,
|
|
pub created_at_utc: String,
|
|
pub reason: BackupReason,
|
|
pub historical_apply: HistoricalApplyStatus,
|
|
pub changes: crate::app::RestoreChangeSummary,
|
|
pub preview_generation: u64,
|
|
}
|
|
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct OperationError {
|
|
pub message: String,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum WorkerEvent {
|
|
SessionReady(SessionView),
|
|
OperationStarted(PendingOperation),
|
|
OperationFinished {
|
|
operation: PendingOperation,
|
|
result: Result<CommandResult, OperationError>,
|
|
view: Box<SessionView>,
|
|
},
|
|
Statistics {
|
|
generation: u64,
|
|
observation: RuntimeObservation,
|
|
},
|
|
ShutdownComplete,
|
|
}
|
|
|
|
/// The UI never shares its Application with the input/render thread. Tests may
|
|
/// inject a delayed/failing executor; production uses `ApplicationExecutor`.
|
|
pub trait OperationExecutor: Send + 'static {
|
|
fn execute(
|
|
&mut self,
|
|
app: &mut Application,
|
|
command: UiCommand,
|
|
) -> Result<CommandResult, OperationError>;
|
|
|
|
fn execute_sensitive(
|
|
&mut self,
|
|
_app: &mut Application,
|
|
_command: SensitiveCommand,
|
|
) -> Result<CommandResult, OperationError> {
|
|
Err(OperationError {
|
|
message: "sensitive operation is unavailable".to_owned(),
|
|
})
|
|
}
|
|
|
|
fn execute_at(
|
|
&mut self,
|
|
app: &mut Application,
|
|
command: UiCommand,
|
|
_generation: u64,
|
|
) -> Result<CommandResult, OperationError> {
|
|
self.execute(app, command)
|
|
}
|
|
fn execute_sensitive_at(
|
|
&mut self,
|
|
app: &mut Application,
|
|
command: SensitiveCommand,
|
|
_generation: u64,
|
|
) -> Result<CommandResult, OperationError> {
|
|
self.execute_sensitive(app, command)
|
|
}
|
|
}
|
|
|
|
enum DeliveryMode {
|
|
FirstIssue,
|
|
Reissue,
|
|
}
|
|
|
|
struct PreparedDestinationContext {
|
|
identity: PublicKey,
|
|
include_qr: bool,
|
|
mode: DeliveryMode,
|
|
attempt: FirstIssueAttempt,
|
|
generation: u64,
|
|
destination: PreparedDestination,
|
|
}
|
|
|
|
struct RestorePreviewContext {
|
|
_archive_name: String,
|
|
generation: u64,
|
|
preview: RestorePreview,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
pub struct ApplicationExecutor {
|
|
prepared_destination: Option<PreparedDestinationContext>,
|
|
restore_preview: Option<RestorePreviewContext>,
|
|
}
|
|
|
|
impl OperationExecutor for ApplicationExecutor {
|
|
fn execute(
|
|
&mut self,
|
|
app: &mut Application,
|
|
command: UiCommand,
|
|
) -> Result<CommandResult, OperationError> {
|
|
self.execute_at(app, command, 0)
|
|
}
|
|
|
|
fn execute_at(
|
|
&mut self,
|
|
app: &mut Application,
|
|
command: UiCommand,
|
|
generation: u64,
|
|
) -> Result<CommandResult, OperationError> {
|
|
let delivery_continuation = matches!(command, UiCommand::DeliverFirstIssue { .. });
|
|
let restore_continuation = matches!(
|
|
command,
|
|
UiCommand::ConfirmRestore { .. } | UiCommand::CancelRestorePreview
|
|
);
|
|
if !delivery_continuation && !matches!(&command, UiCommand::CancelPreparedDelivery) {
|
|
self.prepared_destination = None;
|
|
}
|
|
if !restore_continuation && !matches!(&command, UiCommand::PreviewRestore { .. }) {
|
|
self.restore_preview = None;
|
|
}
|
|
let result = match command {
|
|
UiCommand::CreateClient {
|
|
alias,
|
|
manual_addresses,
|
|
created_utc,
|
|
} => app
|
|
.prepare_new_client(&alias, manual_addresses, created_utc.as_deref())
|
|
.map(CommandResult::ClientCreated),
|
|
UiCommand::RenameClient { identity, alias } => app
|
|
.rename_client(identity, &alias)
|
|
.map(|()| CommandResult::ClientRenamed),
|
|
UiCommand::DeleteClient { identity } => app
|
|
.delete_client(identity)
|
|
.map(|record_removed| CommandResult::ClientDeleted { record_removed }),
|
|
UiCommand::DiscardDraft => app.discard_draft().map(|()| CommandResult::DraftDiscarded),
|
|
UiCommand::Commit => app.commit().map(CommandResult::Committed),
|
|
UiCommand::CreateBackup => app.create_manual_backup().map(CommandResult::BackupCreated),
|
|
UiCommand::ListBackups => app.list_backups().map(CommandResult::Backups),
|
|
UiCommand::ImportSelectedFile { file_name } => app
|
|
.import_selected_file(&file_name)
|
|
.map(CommandResult::SelectedFileImported),
|
|
UiCommand::PreviewRestart => app.preview_restart().map(|preview| {
|
|
CommandResult::RestartPreview(RestartPreviewSummary {
|
|
has_draft: preview.has_draft(),
|
|
interface: preview.interface().to_owned(),
|
|
systemd_unit: preview.systemd_unit().to_owned(),
|
|
})
|
|
}),
|
|
UiCommand::Restart(choice) => {
|
|
let result = app
|
|
.preview_restart()
|
|
.map(|preview| app.restart(preview, choice));
|
|
result.map(CommandResult::Restarted)
|
|
}
|
|
UiCommand::PrepareDelivery {
|
|
identity,
|
|
path,
|
|
include_qr,
|
|
reissue,
|
|
attempt,
|
|
} => app
|
|
.prepare_first_issue_destination(path)
|
|
.map_err(crate::app::AppError::FileOutput)
|
|
.map(|destination| {
|
|
let summary = DeliveryDestinationSummary {
|
|
identity,
|
|
path: destination.path().to_path_buf(),
|
|
overwrite_required: destination.overwrite_confirmation_required(),
|
|
include_qr,
|
|
reissue,
|
|
attempt,
|
|
prepared_generation: generation,
|
|
};
|
|
self.prepared_destination = Some(PreparedDestinationContext {
|
|
identity,
|
|
include_qr,
|
|
mode: if reissue {
|
|
DeliveryMode::Reissue
|
|
} else {
|
|
DeliveryMode::FirstIssue
|
|
},
|
|
attempt,
|
|
generation,
|
|
destination,
|
|
});
|
|
CommandResult::DeliveryDestination(summary)
|
|
}),
|
|
UiCommand::DeliverFirstIssue {
|
|
attempt,
|
|
overwrite_confirmed,
|
|
prepared_generation,
|
|
} => {
|
|
let prepared = self
|
|
.prepared_destination
|
|
.take()
|
|
.ok_or_else(|| OperationError {
|
|
message: "No prepared output destination is available.".to_owned(),
|
|
})?;
|
|
if !matches!(prepared.mode, DeliveryMode::FirstIssue)
|
|
|| prepared.attempt != attempt
|
|
|| !destination_generation_matches(
|
|
prepared.generation,
|
|
prepared_generation,
|
|
generation,
|
|
)
|
|
{
|
|
return Err(OperationError {
|
|
message: "Prepared output destination is stale; select it again."
|
|
.to_owned(),
|
|
});
|
|
}
|
|
let identity = prepared.identity;
|
|
let include_qr = prepared.include_qr;
|
|
let mut destination = prepared.destination;
|
|
if destination.overwrite_confirmation_required() {
|
|
if !overwrite_confirmed {
|
|
return Err(OperationError {
|
|
message: "Overwrite confirmation was not supplied.".to_owned(),
|
|
});
|
|
}
|
|
destination = destination.confirm_overwrite().map_err(operation_error)?;
|
|
}
|
|
if include_qr {
|
|
app.deliver_first_issue_file_and_qr(identity, attempt, destination)
|
|
.map(|outcome| match outcome {
|
|
FirstIssueDeliveryOutcome::Delivered(delivery) => {
|
|
let (qr, qr_output) = qr_result(delivery.qr);
|
|
CommandResult::FirstIssue(FirstIssueSummary {
|
|
identity,
|
|
status: FirstIssueStatus::Delivered,
|
|
retry: None,
|
|
activation_warning: delivery.activation_warning,
|
|
qr,
|
|
qr_output,
|
|
})
|
|
}
|
|
FirstIssueDeliveryOutcome::PublishedButBookkeepingUncertain {
|
|
value,
|
|
..
|
|
} => {
|
|
let (qr, qr_output) = qr_result(value.qr);
|
|
CommandResult::FirstIssue(FirstIssueSummary {
|
|
identity,
|
|
status: FirstIssueStatus::PublishedButBookkeepingUncertain,
|
|
retry: None,
|
|
activation_warning: value.activation_warning,
|
|
qr,
|
|
qr_output,
|
|
})
|
|
}
|
|
FirstIssueDeliveryOutcome::Failed {
|
|
error,
|
|
retry,
|
|
activation_warning,
|
|
} => CommandResult::FirstIssue(FirstIssueSummary {
|
|
identity,
|
|
status: FirstIssueStatus::Failed(error),
|
|
retry: Some(retry),
|
|
activation_warning,
|
|
qr: QrStatus::NotRequested,
|
|
qr_output: None,
|
|
}),
|
|
})
|
|
} else {
|
|
app.deliver_first_issue_file(identity, attempt, destination)
|
|
.map(|outcome| match outcome {
|
|
FirstIssueDeliveryOutcome::Delivered(delivery) => {
|
|
CommandResult::FirstIssue(FirstIssueSummary {
|
|
identity,
|
|
status: FirstIssueStatus::Delivered,
|
|
retry: None,
|
|
activation_warning: delivery.activation_warning,
|
|
qr: QrStatus::NotRequested,
|
|
qr_output: None,
|
|
})
|
|
}
|
|
FirstIssueDeliveryOutcome::PublishedButBookkeepingUncertain {
|
|
value,
|
|
..
|
|
} => CommandResult::FirstIssue(FirstIssueSummary {
|
|
identity,
|
|
status: FirstIssueStatus::PublishedButBookkeepingUncertain,
|
|
retry: None,
|
|
activation_warning: value.activation_warning,
|
|
qr: QrStatus::NotRequested,
|
|
qr_output: None,
|
|
}),
|
|
FirstIssueDeliveryOutcome::Failed {
|
|
error,
|
|
retry,
|
|
activation_warning,
|
|
} => CommandResult::FirstIssue(FirstIssueSummary {
|
|
identity,
|
|
status: FirstIssueStatus::Failed(error),
|
|
retry: Some(retry),
|
|
activation_warning,
|
|
qr: QrStatus::NotRequested,
|
|
qr_output: None,
|
|
}),
|
|
})
|
|
}
|
|
}
|
|
UiCommand::PreviewRestore { file_name } => {
|
|
app.preview_restore(&file_name).map(|preview| {
|
|
let changes = preview.change_summary(app.saved_bytes());
|
|
let summary = RestorePreviewSummary {
|
|
file_name: file_name.clone(),
|
|
created_at_utc: preview.created_at_utc().to_owned(),
|
|
reason: preview.reason(),
|
|
historical_apply: preview.historical_apply().clone(),
|
|
changes,
|
|
preview_generation: generation,
|
|
};
|
|
self.restore_preview = Some(RestorePreviewContext {
|
|
_archive_name: file_name,
|
|
generation,
|
|
preview,
|
|
});
|
|
CommandResult::RestorePreview(summary)
|
|
})
|
|
}
|
|
UiCommand::CancelPreparedDelivery => {
|
|
self.prepared_destination = None;
|
|
Ok(CommandResult::Cancelled)
|
|
}
|
|
UiCommand::CancelRestorePreview => {
|
|
self.restore_preview = None;
|
|
Ok(CommandResult::Cancelled)
|
|
}
|
|
UiCommand::ConfirmRestore { preview_generation } => {
|
|
let context = self.restore_preview.take().ok_or_else(|| OperationError {
|
|
message: "Restore preview is no longer available; select the archive again."
|
|
.to_owned(),
|
|
})?;
|
|
if !restore_preview_generation_matches(
|
|
context.generation,
|
|
preview_generation,
|
|
generation,
|
|
) {
|
|
return Err(OperationError {
|
|
message: "Restore preview is stale; select the archive again.".to_owned(),
|
|
});
|
|
}
|
|
app.confirm_restore(context.preview)
|
|
.map(CommandResult::Restored)
|
|
}
|
|
};
|
|
result.map_err(|error| OperationError {
|
|
message: error.to_string(),
|
|
})
|
|
}
|
|
|
|
fn execute_sensitive(
|
|
&mut self,
|
|
app: &mut Application,
|
|
command: SensitiveCommand,
|
|
) -> Result<CommandResult, OperationError> {
|
|
match command {
|
|
SensitiveCommand::PrepareInitialSetup(password) => app
|
|
.prepare_initial_setup_from_sources(password.expose())
|
|
.map(CommandResult::InitialSetupPrepared)
|
|
.map_err(|error| OperationError {
|
|
message: error.to_string(),
|
|
}),
|
|
SensitiveCommand::ReissueClient {
|
|
identity,
|
|
password,
|
|
overwrite_confirmed,
|
|
prepared_generation,
|
|
} => {
|
|
let prepared = self
|
|
.prepared_destination
|
|
.take()
|
|
.ok_or_else(|| OperationError {
|
|
message: "No prepared output destination is available.".to_owned(),
|
|
})?;
|
|
if prepared.identity != identity
|
|
|| !matches!(prepared.mode, DeliveryMode::Reissue)
|
|
|| prepared.generation != prepared_generation
|
|
{
|
|
return Err(OperationError {
|
|
message: "Prepared output destination is stale; select it again."
|
|
.to_owned(),
|
|
});
|
|
}
|
|
let include_qr = prepared.include_qr;
|
|
let mut destination = prepared.destination;
|
|
if destination.overwrite_confirmation_required() {
|
|
if !overwrite_confirmed {
|
|
return Err(OperationError {
|
|
message: "Overwrite confirmation was not supplied.".to_owned(),
|
|
});
|
|
}
|
|
destination = destination.confirm_overwrite().map_err(operation_error)?;
|
|
}
|
|
let delivery = app
|
|
.reissue_client_file(identity, password.expose(), destination, include_qr)
|
|
.map_err(|error| OperationError {
|
|
message: error.to_string(),
|
|
})?;
|
|
let (qr, qr_output) = match delivery.qr {
|
|
None => (QrStatus::NotRequested, None),
|
|
Some(result) => qr_result(result),
|
|
};
|
|
Ok(CommandResult::Reissued(ReissueSummary {
|
|
activation_warning: delivery.activation_warning,
|
|
qr,
|
|
qr_output,
|
|
}))
|
|
}
|
|
}
|
|
}
|
|
|
|
fn execute_sensitive_at(
|
|
&mut self,
|
|
app: &mut Application,
|
|
command: SensitiveCommand,
|
|
generation: u64,
|
|
) -> Result<CommandResult, OperationError> {
|
|
match &command {
|
|
SensitiveCommand::ReissueClient {
|
|
prepared_generation,
|
|
..
|
|
} if *prepared_generation != generation => {
|
|
return Err(OperationError {
|
|
message: "Prepared output destination is stale; select it again.".to_owned(),
|
|
});
|
|
}
|
|
_ => {}
|
|
}
|
|
if matches!(command, SensitiveCommand::PrepareInitialSetup(_)) {
|
|
self.prepared_destination = None;
|
|
self.restore_preview = None;
|
|
}
|
|
self.execute_sensitive(app, command)
|
|
}
|
|
}
|
|
|
|
fn operation_error(error: impl std::fmt::Display) -> OperationError {
|
|
OperationError {
|
|
message: error.to_string(),
|
|
}
|
|
}
|
|
|
|
fn destination_generation_matches(stored: u64, requested: u64, current: u64) -> bool {
|
|
stored == requested && stored == current
|
|
}
|
|
|
|
fn restore_preview_generation_matches(stored: u64, requested: u64, current: u64) -> bool {
|
|
stored == requested && current == requested.saturating_add(1)
|
|
}
|
|
|
|
fn qr_result(result: Result<QrCodeOutput, QrError>) -> (QrStatus, Option<QrCodeOutput>) {
|
|
match result {
|
|
Ok(output) => (QrStatus::Ready, Some(output)),
|
|
Err(error) => (QrStatus::Failed(error), None),
|
|
}
|
|
}
|
|
|
|
pub trait StatisticsReader: Send + 'static {
|
|
/// Implementations must return within `timeout`; unavailable runtime data
|
|
/// is an observation, not a failure or inferred offline status.
|
|
fn read(&mut self, interface: &str, timeout: Duration) -> RuntimeObservation;
|
|
}
|
|
|
|
pub struct UnavailableStatistics;
|
|
|
|
impl StatisticsReader for UnavailableStatistics {
|
|
fn read(&mut self, _interface: &str, _timeout: Duration) -> RuntimeObservation {
|
|
RuntimeObservation {
|
|
observed_at: None,
|
|
interface: crate::runtime::InterfaceObservation::Unavailable,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Read-only Linux WireGuard observation adapter. Failures remain unavailable
|
|
/// observations and never alter the application's applied-state marker.
|
|
pub struct LinuxStatisticsReader;
|
|
|
|
impl StatisticsReader for LinuxStatisticsReader {
|
|
fn read(&mut self, interface: &str, timeout: Duration) -> RuntimeObservation {
|
|
match crate::runtime::transport::observe_interface(
|
|
interface,
|
|
timeout,
|
|
crate::runtime::netlink::DecodeLimits::default(),
|
|
) {
|
|
Ok(observation) => observation,
|
|
Err(crate::runtime::transport::TransportError::NoDevice) => RuntimeObservation {
|
|
observed_at: None,
|
|
interface: crate::runtime::InterfaceObservation::Missing,
|
|
},
|
|
Err(_) => RuntimeObservation {
|
|
observed_at: None,
|
|
interface: crate::runtime::InterfaceObservation::Unavailable,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
pub struct WorkerConfig {
|
|
pub refresh_interval: Duration,
|
|
pub read_timeout: Duration,
|
|
}
|
|
|
|
impl Default for WorkerConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
refresh_interval: DEFAULT_REFRESH_INTERVAL,
|
|
read_timeout: DEFAULT_READ_TIMEOUT,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
pub enum SubmitError {
|
|
Busy,
|
|
QueueFull,
|
|
ShuttingDown,
|
|
RequestIdsExhausted,
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
pub enum ShutdownProgress {
|
|
Pending,
|
|
Stopped,
|
|
Disconnected,
|
|
}
|
|
|
|
enum WorkerMessage {
|
|
Execute(PendingOperation, WorkerCommand),
|
|
Refresh,
|
|
Shutdown,
|
|
}
|
|
|
|
enum WorkerCommand {
|
|
Public(UiCommand),
|
|
Sensitive(SensitiveCommand),
|
|
}
|
|
|
|
/// Handle stays live while bounded shutdown is pending. Call `shutdown` again
|
|
/// to collect the final operation result and join only after its completion
|
|
/// event; never discard the handle while a protected operation is submitted.
|
|
#[must_use = "retain the worker and keep rendering progress until shutdown completes"]
|
|
pub struct ApplicationWorker {
|
|
sender: SyncSender<WorkerMessage>,
|
|
events: Receiver<WorkerEvent>,
|
|
pending_events: VecDeque<WorkerEvent>,
|
|
busy: Arc<AtomicBool>,
|
|
stop_requested: Arc<AtomicBool>,
|
|
shutting_down: bool,
|
|
shutdown_complete: bool,
|
|
next_request_id: AtomicU64,
|
|
generation: Arc<AtomicU64>,
|
|
refresh_pending: Arc<AtomicBool>,
|
|
join: Option<JoinHandle<()>>,
|
|
}
|
|
|
|
impl ApplicationWorker {
|
|
pub fn spawn(
|
|
app: Application,
|
|
statistics: impl StatisticsReader,
|
|
config: WorkerConfig,
|
|
) -> std::io::Result<Self> {
|
|
Self::spawn_with_executor(app, ApplicationExecutor::default(), statistics, config)
|
|
}
|
|
|
|
pub fn spawn_with_executor(
|
|
app: Application,
|
|
executor: impl OperationExecutor,
|
|
statistics: impl StatisticsReader,
|
|
config: WorkerConfig,
|
|
) -> std::io::Result<Self> {
|
|
let (sender, receiver) = mpsc::sync_channel(COMMAND_QUEUE_CAPACITY);
|
|
let (event_sender, events) = mpsc::sync_channel(EVENT_QUEUE_CAPACITY);
|
|
let busy = Arc::new(AtomicBool::new(false));
|
|
let worker_busy = Arc::clone(&busy);
|
|
let stop_requested = Arc::new(AtomicBool::new(false));
|
|
let worker_stop_requested = Arc::clone(&stop_requested);
|
|
let generation = Arc::new(AtomicU64::new(0));
|
|
let worker_generation = Arc::clone(&generation);
|
|
let refresh_pending = Arc::new(AtomicBool::new(false));
|
|
let worker_refresh_pending = Arc::clone(&refresh_pending);
|
|
let join = thread::Builder::new()
|
|
.name("wireguard-admin-app".to_owned())
|
|
.spawn(move || {
|
|
run_worker(
|
|
app,
|
|
WorkerLoop {
|
|
executor: Box::new(executor),
|
|
statistics: Box::new(statistics),
|
|
receiver,
|
|
events: event_sender,
|
|
busy: worker_busy,
|
|
stop_requested: worker_stop_requested,
|
|
generation: worker_generation,
|
|
refresh_pending: worker_refresh_pending,
|
|
config,
|
|
},
|
|
);
|
|
})?;
|
|
Ok(Self {
|
|
sender,
|
|
events,
|
|
pending_events: VecDeque::with_capacity(EVENT_QUEUE_CAPACITY),
|
|
busy,
|
|
stop_requested,
|
|
shutting_down: false,
|
|
shutdown_complete: false,
|
|
next_request_id: AtomicU64::new(1),
|
|
generation,
|
|
refresh_pending,
|
|
join: Some(join),
|
|
})
|
|
}
|
|
|
|
pub fn submit(&self, command: UiCommand) -> Result<PendingOperation, SubmitError> {
|
|
let kind = command.kind();
|
|
let changes_view = command.changes_view();
|
|
self.submit_request(WorkerCommand::Public(command), kind, changes_view)
|
|
}
|
|
|
|
pub fn submit_sensitive(
|
|
&self,
|
|
command: SensitiveCommand,
|
|
) -> Result<PendingOperation, SubmitError> {
|
|
let kind = command.kind();
|
|
let changes_view = command.changes_view();
|
|
self.submit_request(WorkerCommand::Sensitive(command), kind, changes_view)
|
|
}
|
|
|
|
fn submit_request(
|
|
&self,
|
|
command: WorkerCommand,
|
|
kind: OperationKind,
|
|
changes_view: bool,
|
|
) -> Result<PendingOperation, SubmitError> {
|
|
if self.shutting_down {
|
|
return Err(SubmitError::ShuttingDown);
|
|
}
|
|
if self
|
|
.busy
|
|
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
|
.is_err()
|
|
{
|
|
return Err(SubmitError::Busy);
|
|
}
|
|
let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed);
|
|
if request_id == u64::MAX {
|
|
self.busy.store(false, Ordering::Release);
|
|
return Err(SubmitError::RequestIdsExhausted);
|
|
}
|
|
let generation = if changes_view {
|
|
self.generation
|
|
.fetch_add(1, Ordering::AcqRel)
|
|
.saturating_add(1)
|
|
} else {
|
|
self.generation.load(Ordering::Acquire)
|
|
};
|
|
let operation = PendingOperation {
|
|
request_id,
|
|
kind,
|
|
generation,
|
|
};
|
|
match self
|
|
.sender
|
|
.try_send(WorkerMessage::Execute(operation, command))
|
|
{
|
|
Ok(()) => Ok(operation),
|
|
Err(TrySendError::Full(_)) => {
|
|
self.busy.store(false, Ordering::Release);
|
|
Err(SubmitError::QueueFull)
|
|
}
|
|
Err(TrySendError::Disconnected(_)) => {
|
|
self.busy.store(false, Ordering::Release);
|
|
Err(SubmitError::ShuttingDown)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Coalesce refresh requests. Periodic refresh also runs at the configured
|
|
/// interval; neither path can queue an unbounded number of samples.
|
|
pub fn request_statistics(&self) -> Result<(), SubmitError> {
|
|
if self.shutting_down {
|
|
return Err(SubmitError::ShuttingDown);
|
|
}
|
|
if self.refresh_pending.swap(true, Ordering::AcqRel) {
|
|
return Ok(());
|
|
}
|
|
match self.sender.try_send(WorkerMessage::Refresh) {
|
|
Ok(()) => Ok(()),
|
|
Err(TrySendError::Full(_)) => Ok(()),
|
|
Err(TrySendError::Disconnected(_)) => Err(SubmitError::ShuttingDown),
|
|
}
|
|
}
|
|
|
|
pub fn try_receive(&mut self) -> Option<WorkerEvent> {
|
|
self.pending_events
|
|
.pop_front()
|
|
.or_else(|| self.events.try_recv().ok())
|
|
}
|
|
|
|
pub fn current_generation(&self) -> u64 {
|
|
self.generation.load(Ordering::Acquire)
|
|
}
|
|
|
|
/// Request shutdown and wait only for the supplied bound. `Pending` means
|
|
/// an accepted operation may still be saving or restarting; keep the UI and
|
|
/// this handle alive and call again to observe its result.
|
|
pub fn shutdown(&mut self, timeout: Duration) -> ShutdownProgress {
|
|
if self.shutdown_complete {
|
|
return ShutdownProgress::Stopped;
|
|
}
|
|
if !self.shutting_down {
|
|
self.shutting_down = true;
|
|
self.stop_requested.store(true, Ordering::Release);
|
|
// The atomic flag wakes the worker's drain logic even when a full
|
|
// bounded queue cannot accept a shutdown message.
|
|
let _ = self.sender.try_send(WorkerMessage::Shutdown);
|
|
}
|
|
let deadline = Instant::now() + timeout;
|
|
loop {
|
|
let remaining = deadline.saturating_duration_since(Instant::now());
|
|
if remaining.is_zero() {
|
|
return ShutdownProgress::Pending;
|
|
}
|
|
match self.events.recv_timeout(remaining) {
|
|
Ok(WorkerEvent::ShutdownComplete) => {
|
|
self.shutdown_complete = true;
|
|
if let Some(join) = self.join.take() {
|
|
let _ = join.join();
|
|
}
|
|
return ShutdownProgress::Stopped;
|
|
}
|
|
Ok(event) => self.retain_pending_event(event),
|
|
Err(RecvTimeoutError::Timeout) => return ShutdownProgress::Pending,
|
|
Err(RecvTimeoutError::Disconnected) => return ShutdownProgress::Disconnected,
|
|
}
|
|
}
|
|
}
|
|
|
|
fn retain_pending_event(&mut self, event: WorkerEvent) {
|
|
if self.pending_events.len() < EVENT_QUEUE_CAPACITY {
|
|
self.pending_events.push_back(event);
|
|
return;
|
|
}
|
|
if let Some(index) = self
|
|
.pending_events
|
|
.iter()
|
|
.position(|queued| matches!(queued, WorkerEvent::Statistics { .. }))
|
|
{
|
|
self.pending_events.remove(index);
|
|
self.pending_events.push_back(event);
|
|
}
|
|
}
|
|
}
|
|
|
|
struct WorkerLoop {
|
|
executor: Box<dyn OperationExecutor>,
|
|
statistics: Box<dyn StatisticsReader>,
|
|
receiver: Receiver<WorkerMessage>,
|
|
events: SyncSender<WorkerEvent>,
|
|
busy: Arc<AtomicBool>,
|
|
stop_requested: Arc<AtomicBool>,
|
|
generation: Arc<AtomicU64>,
|
|
refresh_pending: Arc<AtomicBool>,
|
|
config: WorkerConfig,
|
|
}
|
|
|
|
fn run_worker(mut app: Application, worker: WorkerLoop) {
|
|
let WorkerLoop {
|
|
mut executor,
|
|
mut statistics,
|
|
receiver,
|
|
events,
|
|
busy,
|
|
stop_requested,
|
|
generation,
|
|
refresh_pending,
|
|
config,
|
|
} = worker;
|
|
let mut next_refresh = Instant::now() + config.refresh_interval;
|
|
let initial_view = SessionView::from_application(&app, generation.load(Ordering::Acquire));
|
|
if events
|
|
.send(WorkerEvent::SessionReady(initial_view))
|
|
.is_err()
|
|
{
|
|
return;
|
|
}
|
|
loop {
|
|
let now = Instant::now();
|
|
let until_refresh = next_refresh.saturating_duration_since(now);
|
|
let message = if stop_requested.load(Ordering::Acquire) {
|
|
match receiver.try_recv() {
|
|
Ok(message) => Some(message),
|
|
Err(mpsc::TryRecvError::Empty | mpsc::TryRecvError::Disconnected) => break,
|
|
}
|
|
} else if refresh_pending.swap(false, Ordering::AcqRel) || until_refresh.is_zero() {
|
|
None
|
|
} else {
|
|
match receiver.recv_timeout(until_refresh) {
|
|
Ok(message) => Some(message),
|
|
Err(RecvTimeoutError::Timeout) => None,
|
|
Err(RecvTimeoutError::Disconnected) => break,
|
|
}
|
|
};
|
|
match message {
|
|
Some(WorkerMessage::Execute(operation, command)) => {
|
|
let _ = events.send(WorkerEvent::OperationStarted(operation));
|
|
let result = match command {
|
|
WorkerCommand::Public(command) => {
|
|
executor.execute_at(&mut app, command, operation.generation)
|
|
}
|
|
WorkerCommand::Sensitive(command) => {
|
|
executor.execute_sensitive_at(&mut app, command, operation.generation)
|
|
}
|
|
};
|
|
let view = SessionView::from_application(&app, operation.generation);
|
|
if events
|
|
.send(WorkerEvent::OperationFinished {
|
|
operation,
|
|
result,
|
|
view: Box::new(view),
|
|
})
|
|
.is_err()
|
|
{
|
|
break;
|
|
}
|
|
busy.store(false, Ordering::Release);
|
|
}
|
|
Some(WorkerMessage::Refresh) => {}
|
|
Some(WorkerMessage::Shutdown) => break,
|
|
None => {
|
|
let before = generation.load(Ordering::Acquire);
|
|
let interface = std::str::from_utf8(&app.saved_bytes().settings)
|
|
.ok()
|
|
.and_then(|text| SettingsFile::parse_toml(text).ok())
|
|
.and_then(|settings| settings.validate().ok())
|
|
.map(|settings| settings.interface_name);
|
|
let observation = interface.map_or_else(RuntimeObservation::unknown, |interface| {
|
|
statistics.read(&interface, config.read_timeout)
|
|
});
|
|
let after = generation.load(Ordering::Acquire);
|
|
if before == after {
|
|
let _ = events.try_send(WorkerEvent::Statistics {
|
|
generation: after,
|
|
observation,
|
|
});
|
|
}
|
|
next_refresh = Instant::now() + config.refresh_interval;
|
|
}
|
|
}
|
|
}
|
|
let _ = events.send(WorkerEvent::ShutdownComplete);
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use std::fs;
|
|
use std::net::{Ipv4Addr, Ipv6Addr};
|
|
use std::os::unix::fs::PermissionsExt;
|
|
use std::sync::atomic::AtomicUsize;
|
|
use std::sync::mpsc::{self, Receiver};
|
|
|
|
use crate::config::SettingsFile;
|
|
use crate::domain::Client;
|
|
use crate::persistence::SavedStore;
|
|
use crate::vault::WireGuardPrivateKey;
|
|
|
|
#[test]
|
|
fn prepared_delivery_and_restore_confirmations_are_bound_to_the_reviewed_generation() {
|
|
assert!(destination_generation_matches(8, 8, 8));
|
|
assert!(!destination_generation_matches(8, 9, 8));
|
|
assert!(!destination_generation_matches(8, 8, 9));
|
|
assert!(restore_preview_generation_matches(8, 8, 9));
|
|
assert!(!restore_preview_generation_matches(7, 8, 9));
|
|
assert!(!restore_preview_generation_matches(8, 8, 8));
|
|
}
|
|
|
|
#[test]
|
|
fn prepared_delivery_is_cleared_by_cancel_and_unrelated_commands() {
|
|
let (_root, mut app) = application();
|
|
let identity = PublicKey::from_bytes([44; 32]);
|
|
let mut executor = ApplicationExecutor::default();
|
|
let path = _root.path().join("prepared.conf");
|
|
let prepare = || UiCommand::PrepareDelivery {
|
|
identity,
|
|
path: path.clone(),
|
|
include_qr: false,
|
|
reissue: false,
|
|
attempt: FirstIssueAttempt::Initial,
|
|
};
|
|
let deliver = || UiCommand::DeliverFirstIssue {
|
|
attempt: FirstIssueAttempt::Initial,
|
|
overwrite_confirmed: false,
|
|
prepared_generation: 7,
|
|
};
|
|
|
|
assert!(matches!(
|
|
executor.execute_at(&mut app, prepare(), 7),
|
|
Ok(CommandResult::DeliveryDestination(_))
|
|
));
|
|
assert!(matches!(
|
|
executor.execute_at(&mut app, UiCommand::CancelPreparedDelivery, 7),
|
|
Ok(CommandResult::Cancelled)
|
|
));
|
|
assert_eq!(
|
|
executor
|
|
.execute_at(&mut app, deliver(), 7)
|
|
.unwrap_err()
|
|
.message,
|
|
"No prepared output destination is available."
|
|
);
|
|
|
|
assert!(matches!(
|
|
executor.execute_at(&mut app, prepare(), 7),
|
|
Ok(CommandResult::DeliveryDestination(_))
|
|
));
|
|
assert!(matches!(
|
|
executor.execute_at(&mut app, UiCommand::ListBackups, 7),
|
|
Ok(CommandResult::Backups(_))
|
|
));
|
|
assert_eq!(
|
|
executor
|
|
.execute_at(&mut app, deliver(), 7)
|
|
.unwrap_err()
|
|
.message,
|
|
"No prepared output destination is available."
|
|
);
|
|
}
|
|
|
|
struct FakeStatistics {
|
|
started: Option<mpsc::SyncSender<()>>,
|
|
release: Option<Receiver<()>>,
|
|
calls: Arc<AtomicUsize>,
|
|
}
|
|
|
|
impl StatisticsReader for FakeStatistics {
|
|
fn read(&mut self, _interface: &str, _timeout: Duration) -> RuntimeObservation {
|
|
self.calls.fetch_add(1, Ordering::SeqCst);
|
|
if let Some(started) = self.started.take() {
|
|
let _ = started.send(());
|
|
}
|
|
if let Some(release) = self.release.take() {
|
|
let _ = release.recv();
|
|
}
|
|
RuntimeObservation::unknown()
|
|
}
|
|
}
|
|
|
|
struct FakeExecutor {
|
|
started: mpsc::SyncSender<()>,
|
|
release: Receiver<()>,
|
|
calls: Arc<AtomicUsize>,
|
|
}
|
|
|
|
impl OperationExecutor for FakeExecutor {
|
|
fn execute(
|
|
&mut self,
|
|
_app: &mut Application,
|
|
command: UiCommand,
|
|
) -> Result<CommandResult, OperationError> {
|
|
self.calls.fetch_add(1, Ordering::SeqCst);
|
|
let _ = self.started.send(());
|
|
let _ = self.release.recv();
|
|
if matches!(command, UiCommand::Commit) {
|
|
Err(OperationError {
|
|
message: "synthetic commit failure".to_owned(),
|
|
})
|
|
} else {
|
|
Ok(CommandResult::DraftDiscarded)
|
|
}
|
|
}
|
|
}
|
|
|
|
fn application() -> (tempfile::TempDir, Application) {
|
|
let root = tempfile::tempdir().unwrap();
|
|
let settings_text = r#"interface_name = "wg0"
|
|
server_config_path = "wg0.conf"
|
|
client_source_directory = "clients"
|
|
server_endpoint = "vpn.example.org:51820"
|
|
ipv4_allocation_range = "10.8.0.0/24"
|
|
ipv6_allocation_range = "fd80::/64"
|
|
dns_addresses = ["10.8.0.1"]
|
|
systemd_unit = "wg-quick@wg0.service"
|
|
restart_timeout_seconds = 10
|
|
"#;
|
|
let settings = SettingsFile::parse_toml(settings_text)
|
|
.unwrap()
|
|
.validate()
|
|
.unwrap()
|
|
.resolve_paths(root.path());
|
|
fs::write(
|
|
root.path().join(crate::config::SETTINGS_FILE_NAME),
|
|
settings_text,
|
|
)
|
|
.unwrap();
|
|
let server_key = WireGuardPrivateKey::generate().unwrap();
|
|
let client_key = WireGuardPrivateKey::generate().unwrap();
|
|
let client = Client::new(
|
|
client_key.public_key(),
|
|
"fixture",
|
|
ClientAddresses {
|
|
ipv4: Ipv4Addr::new(10, 8, 0, 2),
|
|
ipv6: Some("fd80::2".parse::<Ipv6Addr>().unwrap()),
|
|
},
|
|
)
|
|
.unwrap();
|
|
let server = format!(
|
|
"[Interface]\nPrivateKey = {}\nAddress = 10.8.0.1/24, fd80::1/64\n\n[Peer]\nPublicKey = {}\nAllowedIPs = 10.8.0.2/32, fd80::2/128\n",
|
|
server_key.to_base64(),
|
|
client.identity
|
|
);
|
|
let server_path = root.path().join("wg0.conf");
|
|
fs::write(&server_path, server).unwrap();
|
|
fs::set_permissions(&server_path, fs::Permissions::from_mode(0o600)).unwrap();
|
|
let store = SavedStore::new(root.path(), settings).unwrap();
|
|
let app = Application::open_from_saved_store(store).unwrap();
|
|
(root, app)
|
|
}
|
|
|
|
#[test]
|
|
fn command_execution_is_serialized_and_delayed_work_keeps_ui_submit_nonblocking() {
|
|
let (started_tx, started_rx) = mpsc::sync_channel(1);
|
|
let (release_tx, release_rx) = mpsc::sync_channel(1);
|
|
let calls = Arc::new(AtomicUsize::new(0));
|
|
let executor = FakeExecutor {
|
|
started: started_tx,
|
|
release: release_rx,
|
|
calls: Arc::clone(&calls),
|
|
};
|
|
let (_root, app) = application();
|
|
let mut worker = ApplicationWorker::spawn_with_executor(
|
|
app,
|
|
executor,
|
|
UnavailableStatistics,
|
|
WorkerConfig {
|
|
refresh_interval: Duration::from_secs(60),
|
|
read_timeout: Duration::from_millis(20),
|
|
},
|
|
)
|
|
.unwrap();
|
|
let operation = worker.submit(UiCommand::Commit).unwrap();
|
|
started_rx.recv_timeout(Duration::from_secs(1)).unwrap();
|
|
assert_eq!(
|
|
worker.submit(UiCommand::DiscardDraft),
|
|
Err(SubmitError::Busy)
|
|
);
|
|
let started = Instant::now();
|
|
assert!(worker.try_receive().is_some());
|
|
assert!(started.elapsed() < Duration::from_millis(100));
|
|
release_tx.send(()).unwrap();
|
|
let deadline = Instant::now() + Duration::from_secs(1);
|
|
let mut finished = false;
|
|
while Instant::now() < deadline {
|
|
if let Some(WorkerEvent::OperationFinished {
|
|
operation: actual,
|
|
result,
|
|
..
|
|
}) = worker.try_receive()
|
|
{
|
|
assert_eq!(actual, operation);
|
|
assert_eq!(result.unwrap_err().message, "synthetic commit failure");
|
|
finished = true;
|
|
break;
|
|
}
|
|
thread::yield_now();
|
|
}
|
|
assert!(finished);
|
|
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
|
assert_eq!(
|
|
worker.shutdown(Duration::from_secs(1)),
|
|
ShutdownProgress::Stopped
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn shutdown_timeout_reports_pending_until_submitted_operation_finishes() {
|
|
let (started_tx, started_rx) = mpsc::sync_channel(1);
|
|
let (release_tx, release_rx) = mpsc::sync_channel(1);
|
|
let executor = FakeExecutor {
|
|
started: started_tx,
|
|
release: release_rx,
|
|
calls: Arc::new(AtomicUsize::new(0)),
|
|
};
|
|
let (_root, app) = application();
|
|
let mut worker = ApplicationWorker::spawn_with_executor(
|
|
app,
|
|
executor,
|
|
UnavailableStatistics,
|
|
WorkerConfig {
|
|
refresh_interval: Duration::from_secs(60),
|
|
read_timeout: Duration::from_millis(20),
|
|
},
|
|
)
|
|
.unwrap();
|
|
worker.submit(UiCommand::Commit).unwrap();
|
|
started_rx.recv_timeout(Duration::from_secs(1)).unwrap();
|
|
let before = Instant::now();
|
|
assert_eq!(
|
|
worker.shutdown(Duration::from_millis(20)),
|
|
ShutdownProgress::Pending
|
|
);
|
|
assert!(before.elapsed() < Duration::from_millis(250));
|
|
release_tx.send(()).unwrap();
|
|
assert_eq!(
|
|
worker.shutdown(Duration::from_secs(1)),
|
|
ShutdownProgress::Stopped
|
|
);
|
|
let mut saw_completion = false;
|
|
while let Some(event) = worker.try_receive() {
|
|
saw_completion |= matches!(event, WorkerEvent::OperationFinished { .. });
|
|
}
|
|
assert!(
|
|
saw_completion,
|
|
"submitted work outcome must remain observable"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn shutdown_with_full_queue_returns_within_bound_while_statistics_is_delayed() {
|
|
let (started_tx, started_rx) = mpsc::sync_channel(1);
|
|
let (release_tx, release_rx) = mpsc::sync_channel(1);
|
|
let (_root, app) = application();
|
|
let mut worker = ApplicationWorker::spawn(
|
|
app,
|
|
FakeStatistics {
|
|
started: Some(started_tx),
|
|
release: Some(release_rx),
|
|
calls: Arc::new(AtomicUsize::new(0)),
|
|
},
|
|
WorkerConfig {
|
|
refresh_interval: Duration::from_secs(60),
|
|
read_timeout: Duration::from_millis(50),
|
|
},
|
|
)
|
|
.unwrap();
|
|
|
|
worker.request_statistics().unwrap();
|
|
started_rx.recv_timeout(Duration::from_secs(1)).unwrap();
|
|
while worker.sender.try_send(WorkerMessage::Refresh).is_ok() {}
|
|
assert!(matches!(
|
|
worker.sender.try_send(WorkerMessage::Refresh),
|
|
Err(TrySendError::Full(_))
|
|
));
|
|
|
|
let before = Instant::now();
|
|
assert_eq!(
|
|
worker.shutdown(Duration::from_millis(20)),
|
|
ShutdownProgress::Pending
|
|
);
|
|
assert!(before.elapsed() < Duration::from_millis(250));
|
|
|
|
release_tx.send(()).unwrap();
|
|
assert_eq!(
|
|
worker.shutdown(Duration::from_secs(1)),
|
|
ShutdownProgress::Stopped
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn delayed_statistics_are_bounded_and_do_not_advance_generation() {
|
|
let (started_tx, started_rx) = mpsc::sync_channel(1);
|
|
let (release_tx, release_rx) = mpsc::sync_channel(1);
|
|
let calls = Arc::new(AtomicUsize::new(0));
|
|
let (_root, app) = application();
|
|
let mut worker = ApplicationWorker::spawn(
|
|
app,
|
|
FakeStatistics {
|
|
started: Some(started_tx),
|
|
release: Some(release_rx),
|
|
calls: Arc::clone(&calls),
|
|
},
|
|
WorkerConfig {
|
|
refresh_interval: Duration::from_secs(60),
|
|
read_timeout: Duration::from_millis(50),
|
|
},
|
|
)
|
|
.unwrap();
|
|
worker.request_statistics().unwrap();
|
|
started_rx.recv_timeout(Duration::from_secs(1)).unwrap();
|
|
for _ in 0..100 {
|
|
worker.request_statistics().unwrap();
|
|
}
|
|
assert_eq!(worker.current_generation(), 0);
|
|
release_tx.send(()).unwrap();
|
|
let deadline = Instant::now() + Duration::from_secs(1);
|
|
while calls.load(Ordering::SeqCst) < 2 && Instant::now() < deadline {
|
|
thread::yield_now();
|
|
}
|
|
assert_eq!(
|
|
calls.load(Ordering::SeqCst),
|
|
2,
|
|
"100 refresh requests coalesce to one pending read"
|
|
);
|
|
assert_eq!(
|
|
worker.shutdown(Duration::from_secs(1)),
|
|
ShutdownProgress::Stopped
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn statistics_started_before_a_mutation_are_discarded_as_stale() {
|
|
let (stats_started_tx, stats_started_rx) = mpsc::sync_channel(1);
|
|
let (stats_release_tx, stats_release_rx) = mpsc::sync_channel(1);
|
|
let (operation_started_tx, operation_started_rx) = mpsc::sync_channel(1);
|
|
let (operation_release_tx, operation_release_rx) = mpsc::sync_channel(1);
|
|
let (_root, app) = application();
|
|
let mut worker = ApplicationWorker::spawn_with_executor(
|
|
app,
|
|
FakeExecutor {
|
|
started: operation_started_tx,
|
|
release: operation_release_rx,
|
|
calls: Arc::new(AtomicUsize::new(0)),
|
|
},
|
|
FakeStatistics {
|
|
started: Some(stats_started_tx),
|
|
release: Some(stats_release_rx),
|
|
calls: Arc::new(AtomicUsize::new(0)),
|
|
},
|
|
WorkerConfig {
|
|
refresh_interval: Duration::from_secs(60),
|
|
read_timeout: Duration::from_millis(50),
|
|
},
|
|
)
|
|
.unwrap();
|
|
worker.request_statistics().unwrap();
|
|
stats_started_rx
|
|
.recv_timeout(Duration::from_secs(1))
|
|
.unwrap();
|
|
let operation = worker.submit(UiCommand::Commit).unwrap();
|
|
assert_eq!(operation.generation, 1);
|
|
stats_release_tx.send(()).unwrap();
|
|
operation_started_rx
|
|
.recv_timeout(Duration::from_secs(1))
|
|
.unwrap();
|
|
operation_release_tx.send(()).unwrap();
|
|
|
|
let deadline = Instant::now() + Duration::from_secs(1);
|
|
let mut saw_finish = false;
|
|
while Instant::now() < deadline {
|
|
if let Some(event) = worker.try_receive() {
|
|
match event {
|
|
WorkerEvent::Statistics { generation, .. } => {
|
|
assert_ne!(generation, 0, "pre-mutation statistics must be dropped");
|
|
}
|
|
WorkerEvent::OperationFinished {
|
|
operation: finished,
|
|
view,
|
|
..
|
|
} => {
|
|
assert_eq!(finished, operation);
|
|
assert_eq!(view.generation, 1);
|
|
saw_finish = true;
|
|
break;
|
|
}
|
|
WorkerEvent::SessionReady(_)
|
|
| WorkerEvent::OperationStarted(_)
|
|
| WorkerEvent::ShutdownComplete => {}
|
|
}
|
|
} else {
|
|
thread::yield_now();
|
|
}
|
|
}
|
|
assert!(saw_finish);
|
|
assert_eq!(
|
|
worker.shutdown(Duration::from_secs(1)),
|
|
ShutdownProgress::Stopped
|
|
);
|
|
}
|
|
}
|