You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bifromq.apache.org by po...@apache.org on 2026/09/22 02:58:56 UTC
(bifromq-tenon) 02/02: Consolidate the program name and exact version. (#1)
This is an automated email from the ASF dual-hosted git repository.
popduke pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/bifromq-tenon.git
commit 1ccda1f97cc9ff1bb902e3ce4ad6bc4d3c1fcaa1
Author: Gu Jiawei <gu...@foxmail.com>
AuthorDate: Mon Sep 21 19:34:58 2026 -0700
Consolidate the program name and exact version. (#1)
---
src/identifiers/mod.rs | 31 +++++++++++++++
src/identifiers/tests.rs | 35 +++++++++++++++-
src/lib.rs | 4 +-
src/pipeline/plugin/instances/revision_tests.rs | 10 +----
src/pipeline/reconfigure/plan/projection.rs | 19 ++++-----
src/pipeline/reconfigure/revision.rs | 13 +++---
src/runner/http/plugins/mod.rs | 8 +++-
src/runner/pipeline/programs/mod.rs | 39 ++++++++----------
src/runner/pipeline/programs/tests.rs | 52 +++++++++++++++---------
src/runner/pipeline/runtime_resolution/mod.rs | 53 ++++++++++++++-----------
src/runner/plugin/store/mod.rs | 25 +++++++-----
src/runner/plugin/store/tests.rs | 22 +++++++---
src/tenon_document/verified/mod.rs | 36 ++++++++++-------
13 files changed, 220 insertions(+), 127 deletions(-)
diff --git a/src/identifiers/mod.rs b/src/identifiers/mod.rs
index b0f1e90..7335829 100644
--- a/src/identifiers/mod.rs
+++ b/src/identifiers/mod.rs
@@ -398,6 +398,37 @@ impl fmt::Display for ExactVersion {
}
}
+/// Identifies one exact Plugin Program version, shared by any instances using it.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PluginProgramIdentity {
+ program_name: ProgramName,
+ exact_version: ExactVersion,
+}
+
+impl PluginProgramIdentity {
+ /// Creates an identity from already validated components.
+ #[must_use]
+ pub const fn from_parts(program_name: ProgramName, exact_version: ExactVersion) -> Self {
+ Self {
+ program_name,
+ exact_version,
+ }
+ }
+
+ /// Returns the validated Program name.
+ #[must_use]
+ pub const fn program_name(&self) -> &ProgramName {
+ &self.program_name
+ }
+
+ /// Returns the validated exact version.
+ #[must_use]
+ pub const fn exact_version(&self) -> &ExactVersion {
+ &self.exact_version
+ }
+}
+
/// A validated Sink Program type identity.
///
/// The value is always the canonical `<programName>@<exactVersion>` pair. It
diff --git a/src/identifiers/tests.rs b/src/identifiers/tests.rs
index 80c6ccc..abac2b9 100644
--- a/src/identifiers/tests.rs
+++ b/src/identifiers/tests.rs
@@ -18,8 +18,8 @@
*/
use super::{
- ExactVersion, FlowId, IdentifierParseError, PluginInstanceId, ProgramName, SinkContractId,
- TenonDocumentId,
+ ExactVersion, FlowId, IdentifierParseError, PluginInstanceId, PluginProgramIdentity,
+ ProgramName, SinkContractId, TenonDocumentId,
};
use crate::contracts::tenon_document::v1_schema_bytes;
use proptest::prelude::*;
@@ -286,3 +286,34 @@ proptest! {
prop_assert_eq!(parsed_sink_contract_id.to_string(), sink_contract_id);
}
}
+
+#[test]
+fn program_identity_keys_distinguish_names_and_exact_versions() -> io::Result<()> {
+ let identity = |name: &str, version: &str| -> io::Result<PluginProgramIdentity> {
+ Ok(PluginProgramIdentity::from_parts(
+ ProgramName::try_from(name).map_err(io::Error::other)?,
+ ExactVersion::try_from(version).map_err(io::Error::other)?,
+ ))
+ };
+ let first = identity("com.example.source", "1.0.0")?;
+ let next_version = identity("com.example.source", "1.0.1")?;
+ let other_program = identity("com.example.sink", "1.0.0")?;
+ let programs = std::collections::HashMap::from([
+ (first.clone(), 1),
+ (next_version.clone(), 2),
+ (other_program.clone(), 3),
+ ]);
+ assert_eq!(programs.len(), 3);
+ assert_eq!(
+ programs.get(&identity("com.example.source", "1.0.0")?),
+ Some(&1)
+ );
+ assert_eq!(programs.get(&next_version), Some(&2));
+ assert_eq!(programs.get(&other_program), Some(&3));
+ let borrowed: std::collections::HashMap<_, _> = programs.iter().collect();
+ assert_eq!(
+ borrowed.get(&identity("com.example.source", "1.0.0")?),
+ Some(&&1)
+ );
+ Ok(())
+}
diff --git a/src/lib.rs b/src/lib.rs
index eca54b0..85eb6fe 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -28,7 +28,9 @@ use std::error::Error;
use std::process::ExitCode;
pub use config::{HttpTlsConfig, RunnerConfig, ScriptVmLimits, ScriptVmLimitsError};
-pub use identifiers::{ExactVersion, FlowId, PluginInstanceId, ProgramName, TenonDocumentId};
+pub use identifiers::{
+ ExactVersion, FlowId, PluginInstanceId, PluginProgramIdentity, ProgramName, TenonDocumentId,
+};
pub use runner::extensions::{
AllowAll, DocumentProtection, ExecutionDenied, ExecutionPermit, ExecutionPolicy,
ExecutionScope, HttpApiAuthorization, HttpAuthRejection, NoHttpAuth, Plaintext, RunnerHooks,
diff --git a/src/pipeline/plugin/instances/revision_tests.rs b/src/pipeline/plugin/instances/revision_tests.rs
index c4ba2da..3a3d725 100644
--- a/src/pipeline/plugin/instances/revision_tests.rs
+++ b/src/pipeline/plugin/instances/revision_tests.rs
@@ -309,10 +309,7 @@ fn launch_material<'a>(
.ok_or("Instance is missing")?;
let program = revision
.programs()
- .get(&(
- instance.program_name().clone(),
- instance.exact_version().clone(),
- ))
+ .get(instance.program_identity())
.ok_or("Program is missing")?;
let working_directory = root.join("instances").join(id.as_str());
std::fs::create_dir_all(&working_directory)?;
@@ -354,10 +351,7 @@ fn controlled_launch_material<'a>(
.ok_or("Instance is missing")?;
let program = revision
.programs()
- .get(&(
- instance.program_name().clone(),
- instance.exact_version().clone(),
- ))
+ .get(instance.program_identity())
.ok_or("Program is missing")?;
Ok(ControlledPluginLaunch::new(
launch,
diff --git a/src/pipeline/reconfigure/plan/projection.rs b/src/pipeline/reconfigure/plan/projection.rs
index 2634b8a..dd233a0 100644
--- a/src/pipeline/reconfigure/plan/projection.rs
+++ b/src/pipeline/reconfigure/plan/projection.rs
@@ -32,7 +32,7 @@ use std::rc::Rc;
use super::ResourceAction;
use super::identity::{RuntimeResourceIdentityRef, SinkContractIdRef};
-use crate::identifiers::{ExactVersion, FlowId, PluginInstanceId, ProgramName};
+use crate::identifiers::{FlowId, PluginInstanceId, PluginProgramIdentity};
use crate::payload_contract::PayloadContractProjectionMaterial;
use crate::pipeline::reconfigure::error::PipelineReconfigureError;
use crate::pipeline::reconfigure::revision::{PipelineRevision, ProgramRuntime};
@@ -123,8 +123,7 @@ impl RuntimeResourceSpec<'_> {
else {
unreachable!("equal resource identities must have the same resource kind")
};
- instance.program_name() == other_instance.program_name()
- && instance.exact_version() == other_instance.exact_version()
+ instance.program_identity() == other_instance.program_identity()
&& instance.config() == other_instance.config()
&& instance.has_same_startup_settings(other_instance)
&& source_layout == other_source_layout
@@ -503,17 +502,13 @@ fn instance_projection<'a, 'index>(
fn project_programs(
model: &PipelineRevision,
-) -> Result<HashMap<(&ProgramName, &ExactVersion), Rc<ProgramProjection>>, PipelineReconfigureError>
-{
+) -> Result<HashMap<&PluginProgramIdentity, Rc<ProgramProjection>>, PipelineReconfigureError> {
let mut projections = HashMap::new();
projections
.try_reserve(model.programs().len())
.map_err(|_| PipelineReconfigureError::ResourceLimitExceeded)?;
- for ((program_name, exact_version), program) in model.programs() {
- let _ = projections.insert(
- (program_name, exact_version),
- Rc::new(project_program(program)),
- );
+ for (identity, program) in model.programs() {
+ let _ = projections.insert(identity, Rc::new(project_program(program)));
}
Ok(projections)
}
@@ -543,7 +538,7 @@ fn project_program(program: &ProgramRuntime) -> ProgramProjection {
)]
fn project_instances<'a>(
model: &'a PipelineRevision,
- programs: &HashMap<(&'a ProgramName, &'a ExactVersion), Rc<ProgramProjection>>,
+ programs: &HashMap<&'a PluginProgramIdentity, Rc<ProgramProjection>>,
available_cpu_count: NonZeroUsize,
) -> Result<HashMap<&'a PluginInstanceId, InstanceProjection<'a>>, PipelineReconfigureError> {
let mut source_layouts = HashMap::new();
@@ -568,7 +563,7 @@ fn project_instances<'a>(
.map_err(|_| PipelineReconfigureError::ResourceLimitExceeded)?;
for (instance_id, instance) in model.document().plugin_instances() {
let program = programs
- .get(&(instance.program_name(), instance.exact_version()))
+ .get(instance.program_identity())
.expect("a received revision must retain every referenced Program material");
let _ = instances.insert(
instance_id,
diff --git a/src/pipeline/reconfigure/revision.rs b/src/pipeline/reconfigure/revision.rs
index 8a48346..494ef5a 100644
--- a/src/pipeline/reconfigure/revision.rs
+++ b/src/pipeline/reconfigure/revision.rs
@@ -25,7 +25,7 @@
//! It owns no process, Queue, task, or apply state.
use crate::contracts::core::PipelineRevisionPlan;
-use crate::identifiers::{ExactVersion, PluginInstanceId, ProgramName};
+use crate::identifiers::{ExactVersion, PluginInstanceId, PluginProgramIdentity, ProgramName};
use crate::payload_contract::PluginProgramPayloadContract;
use crate::tenon_document::verified::VerifiedTenonDocument;
use std::collections::HashMap;
@@ -37,7 +37,7 @@ pub(crate) struct PipelineRevision {
// Runner hashes authored JSONC bytes, which this semantic Document cannot recover.
document_etag: Box<str>,
document: VerifiedTenonDocument,
- programs: HashMap<(ProgramName, ExactVersion), ProgramRuntime>,
+ programs: HashMap<PluginProgramIdentity, ProgramRuntime>,
}
impl PipelineRevision {
@@ -51,7 +51,7 @@ impl PipelineRevision {
.into_iter()
.map(|program| {
(
- (
+ PluginProgramIdentity::from_parts(
ProgramName::from_verified(program.program_name),
ExactVersion::from_verified(program.exact_version),
),
@@ -76,7 +76,7 @@ impl PipelineRevision {
&self.document
}
- pub(crate) fn programs(&self) -> &HashMap<(ProgramName, ExactVersion), ProgramRuntime> {
+ pub(crate) fn programs(&self) -> &HashMap<PluginProgramIdentity, ProgramRuntime> {
&self.programs
}
@@ -90,10 +90,7 @@ impl PipelineRevision {
) -> &ProgramRuntime {
let instance = &self.document.plugin_instances()[instance_id];
self.programs
- .get(&(
- instance.program_name().clone(),
- instance.exact_version().clone(),
- ))
+ .get(instance.program_identity())
.expect("a received Plugin Instance must retain its Program material")
}
}
diff --git a/src/runner/http/plugins/mod.rs b/src/runner/http/plugins/mod.rs
index 64802cd..6ed4084 100644
--- a/src/runner/http/plugins/mod.rs
+++ b/src/runner/http/plugins/mod.rs
@@ -322,13 +322,17 @@ pub(super) async fn delete_program(
fn program_install_response(
result: Result<PluginProgramInstallResult, PluginOperationFailure>,
) -> Response {
- let (status, (program_name, exact_version)) = match result {
+ let (status, identity) = match result {
Ok(PluginProgramInstallResult::Installed(identity)) => (StatusCode::CREATED, identity),
Ok(PluginProgramInstallResult::Unchanged(identity)) => (StatusCode::NO_CONTENT, identity),
Err(error) => return plugin_operation_failure_response(error),
};
RunnerHttpResponse::empty(status)
- .with_location(format!("/plugins/{program_name}/{exact_version}"))
+ .with_location(format!(
+ "/plugins/{}/{}",
+ identity.program_name(),
+ identity.exact_version()
+ ))
.into_response()
}
diff --git a/src/runner/pipeline/programs/mod.rs b/src/runner/pipeline/programs/mod.rs
index cd8ffca..cc40e7a 100644
--- a/src/runner/pipeline/programs/mod.rs
+++ b/src/runner/pipeline/programs/mod.rs
@@ -24,7 +24,7 @@
//! is dropped; wire material is built on demand rather than cached.
use crate::contracts::core::{PluginInterface as ProtocolPluginInterface, PluginProgramRuntime};
-use crate::identifiers::{ExactVersion, ProgramName};
+use crate::identifiers::PluginProgramIdentity;
use crate::payload_contract::PluginInterface;
use crate::runner::plugin::store::PluginProgramEntry;
use std::collections::HashMap;
@@ -34,22 +34,17 @@ use std::sync::Arc;
/// Owns one reference per exact Program selected for Pipeline runtime material.
#[derive(Debug)]
pub(super) struct PipelineProgramSnapshot {
- programs: HashMap<(ProgramName, ExactVersion), Arc<PluginProgramEntry>>,
+ programs: HashMap<PluginProgramIdentity, Arc<PluginProgramEntry>>,
}
impl PipelineProgramSnapshot {
/// Retains the resolver's already selected, deduplicated Entry references.
pub(super) fn retain<'a>(
- entries: impl IntoIterator<
- Item = (
- (&'a ProgramName, &'a ExactVersion),
- &'a Arc<PluginProgramEntry>,
- ),
- >,
+ entries: impl IntoIterator<Item = (&'a PluginProgramIdentity, &'a Arc<PluginProgramEntry>)>,
) -> Self {
let programs = entries
.into_iter()
- .map(|((name, version), entry)| ((name.clone(), version.clone()), Arc::clone(entry)))
+ .map(|(identity, entry)| (identity.clone(), Arc::clone(entry)))
.collect();
Self { programs }
}
@@ -60,20 +55,18 @@ impl PipelineProgramSnapshot {
let mut runtimes: Vec<_> = self
.programs
.iter()
- .map(
- |((program_name, exact_version), program)| PluginProgramRuntime {
- program_name: program_name.as_str().to_owned(),
- exact_version: exact_version.as_str().to_owned(),
- program_directory: revision_absolute_utf8_path(program.directory()).to_owned(),
- command: program.command().to_vec(),
- plugin_interface: match program.interface() {
- PluginInterface::Source => ProtocolPluginInterface::Source,
- PluginInterface::Sink => ProtocolPluginInterface::Sink,
- PluginInterface::SourceAndSink => ProtocolPluginInterface::SourceAndSink,
- } as i32,
- payload_descriptor_set: program.payload_descriptor_bytes().to_vec(),
- },
- )
+ .map(|(identity, program)| PluginProgramRuntime {
+ program_name: identity.program_name().as_str().to_owned(),
+ exact_version: identity.exact_version().as_str().to_owned(),
+ program_directory: revision_absolute_utf8_path(program.directory()).to_owned(),
+ command: program.command().to_vec(),
+ plugin_interface: match program.interface() {
+ PluginInterface::Source => ProtocolPluginInterface::Source,
+ PluginInterface::Sink => ProtocolPluginInterface::Sink,
+ PluginInterface::SourceAndSink => ProtocolPluginInterface::SourceAndSink,
+ } as i32,
+ payload_descriptor_set: program.payload_descriptor_bytes().to_vec(),
+ })
.collect();
runtimes.sort_unstable_by(|left, right| {
(&left.program_name, &left.exact_version)
diff --git a/src/runner/pipeline/programs/tests.rs b/src/runner/pipeline/programs/tests.rs
index 6c3bbd2..81535f3 100644
--- a/src/runner/pipeline/programs/tests.rs
+++ b/src/runner/pipeline/programs/tests.rs
@@ -19,7 +19,7 @@
use super::PipelineProgramSnapshot;
use crate::contracts::core::PluginInterface as ProtocolPluginInterface;
-use crate::identifiers::{ExactVersion, ProgramName};
+use crate::identifiers::{ExactVersion, PluginProgramIdentity, ProgramName};
use crate::payload_contract::PluginInterface;
use crate::runner::plugin::package::tests::{
equivalent_source_program_package, valid_program_package, valid_source_program_package,
@@ -71,7 +71,7 @@ fn all_interfaces_share_one_sorted_runtime_per_exact_program() -> io::Result<()>
]) {
let selected = identity(&runtime.program_name)?;
let entry = store
- .lookup(&selected.0, &selected.1)
+ .lookup(selected.program_name(), selected.exact_version())
.ok_or_else(|| io::Error::other("selected Program is missing"))?;
assert_eq!(runtime.exact_version, "1.0.0");
assert_eq!(Path::new(&runtime.program_directory), entry.directory());
@@ -107,23 +107,31 @@ fn only_the_last_snapshot_release_allows_selected_program_uninstall() -> io::Res
let source_directory = directory.path().join("com.example.source/1.0.0");
assert!(matches!(
- store.uninstall(&source.0, &source.1),
+ store.uninstall(source.program_name(), source.exact_version()),
Err(PluginStoreError::ProgramInUse),
));
assert!(source_directory.is_dir());
- assert!(store.lookup(&source.0, &source.1).is_some());
+ assert!(
+ store
+ .lookup(source.program_name(), source.exact_version())
+ .is_some()
+ );
assert_eq!(
store
- .uninstall(&sink.0, &sink.1)
+ .uninstall(sink.program_name(), sink.exact_version())
.map_err(io::Error::other)?,
PluginUninstallOutcome::Uninstalled,
);
- assert!(store.lookup(&sink.0, &sink.1).is_none());
+ assert!(
+ store
+ .lookup(sink.program_name(), sink.exact_version())
+ .is_none()
+ );
assert!(!directory.path().join("com.example.sink/1.0.0").exists());
drop(first);
assert!(matches!(
- store.uninstall(&source.0, &source.1),
+ store.uninstall(source.program_name(), source.exact_version()),
Err(PluginStoreError::ProgramInUse),
));
assert_eq!(fs::read(source_directory.join("bin/start"))?, b"program");
@@ -131,11 +139,15 @@ fn only_the_last_snapshot_release_allows_selected_program_uninstall() -> io::Res
drop(second);
assert_eq!(
store
- .uninstall(&source.0, &source.1)
+ .uninstall(source.program_name(), source.exact_version())
.map_err(io::Error::other)?,
PluginUninstallOutcome::Uninstalled,
);
- assert!(store.lookup(&source.0, &source.1).is_none());
+ assert!(
+ store
+ .lookup(source.program_name(), source.exact_version())
+ .is_none()
+ );
assert!(!source_directory.exists());
Ok(())
}
@@ -160,19 +172,23 @@ fn equivalent_install_and_unrelated_changes_preserve_snapshot_material_and_reten
.map_err(io::Error::other)?;
assert_eq!(snapshot.runtimes(), original_runtimes);
assert!(matches!(
- store.uninstall(&source.0, &source.1),
+ store.uninstall(source.program_name(), source.exact_version()),
Err(PluginStoreError::ProgramInUse),
));
drop(snapshot);
assert_eq!(
store
- .uninstall(&source.0, &source.1)
+ .uninstall(source.program_name(), source.exact_version())
.map_err(io::Error::other)?,
PluginUninstallOutcome::Uninstalled,
);
let sink = identity("com.example.sink")?;
- assert!(store.lookup(&sink.0, &sink.1).is_some());
+ assert!(
+ store
+ .lookup(sink.program_name(), sink.exact_version())
+ .is_some()
+ );
Ok(())
}
@@ -183,8 +199,8 @@ fn empty_store() -> io::Result<(TempDir, PluginProgramStore)> {
Ok((directory, store))
}
-fn identity(program_name: &str) -> io::Result<(ProgramName, ExactVersion)> {
- Ok((
+fn identity(program_name: &str) -> io::Result<PluginProgramIdentity> {
+ Ok(PluginProgramIdentity::from_parts(
ProgramName::try_from(program_name.to_owned()).map_err(io::Error::other)?,
ExactVersion::try_from(String::from("1.0.0")).map_err(io::Error::other)?,
))
@@ -192,15 +208,15 @@ fn identity(program_name: &str) -> io::Result<(ProgramName, ExactVersion)> {
fn capture_programs(
store: &PluginProgramStore,
- identities: impl IntoIterator<Item = (ProgramName, ExactVersion)>,
+ identities: impl IntoIterator<Item = PluginProgramIdentity>,
) -> io::Result<PipelineProgramSnapshot> {
let identities: Vec<_> = identities.into_iter().collect();
let programs = identities
.iter()
- .map(|(name, version)| {
+ .map(|identity| {
store
- .lookup(name, version)
- .map(|entry| ((name, version), entry))
+ .lookup(identity.program_name(), identity.exact_version())
+ .map(|entry| (identity, entry))
.ok_or_else(|| io::Error::other("Required Program fixture is absent"))
})
.collect::<io::Result<HashMap<_, _>>>()?;
diff --git a/src/runner/pipeline/runtime_resolution/mod.rs b/src/runner/pipeline/runtime_resolution/mod.rs
index 3cc253b..ad5fc30 100644
--- a/src/runner/pipeline/runtime_resolution/mod.rs
+++ b/src/runner/pipeline/runtime_resolution/mod.rs
@@ -67,41 +67,46 @@ impl RuntimeResolver {
let mut references: HashMap<_, Vec<_>> = HashMap::new();
for (id, instance) in document.plugin_instances() {
references
- .entry((instance.program_name(), instance.exact_version()))
+ .entry(instance.program_identity())
.or_default()
.push(id);
}
let mut references: Vec<_> = references.into_iter().collect();
- references.sort_unstable_by_key(|((name, version), _)| (name.as_str(), version.as_str()));
+ references.sort_unstable_by_key(|(identity, _)| {
+ (
+ identity.program_name().as_str(),
+ identity.exact_version().as_str(),
+ )
+ });
let programs: HashMap<_, _> = references
.into_iter()
- .filter_map(|((name, version), ids)| match store.lookup(name, version) {
- Some(entry) if !entry.platforms().contains(&Platform::CURRENT) => {
- issues.push(RuntimeResolutionIssue::PluginPlatformMismatch {
- program_name: name.clone(),
- exact_version: version.clone(),
- platforms: entry.platforms().into(),
- current_platform: Platform::CURRENT,
- plugin_instance_ids: ids.into_iter().cloned().collect(),
- });
- None
- }
- Some(entry) => Some(((name, version), entry)),
- None => {
- issues.push(RuntimeResolutionIssue::PluginProgramMissing {
- program_name: name.clone(),
- exact_version: version.clone(),
- plugin_instance_ids: ids.into_iter().cloned().collect(),
- });
- None
+ .filter_map(|(identity, ids)| {
+ match store.lookup(identity.program_name(), identity.exact_version()) {
+ Some(entry) if !entry.platforms().contains(&Platform::CURRENT) => {
+ issues.push(RuntimeResolutionIssue::PluginPlatformMismatch {
+ program_name: identity.program_name().clone(),
+ exact_version: identity.exact_version().clone(),
+ platforms: entry.platforms().into(),
+ current_platform: Platform::CURRENT,
+ plugin_instance_ids: ids.into_iter().cloned().collect(),
+ });
+ None
+ }
+ Some(entry) => Some((identity, entry)),
+ None => {
+ issues.push(RuntimeResolutionIssue::PluginProgramMissing {
+ program_name: identity.program_name().clone(),
+ exact_version: identity.exact_version().clone(),
+ plugin_instance_ids: ids.into_iter().cloned().collect(),
+ });
+ None
+ }
}
})
.collect();
let program_for_instance = |id: &PluginInstanceId| {
let instance = &document.plugin_instances()[id];
- programs
- .get(&(instance.program_name(), instance.exact_version()))
- .copied()
+ programs.get(instance.program_identity()).copied()
};
// These borrowed groups exist only during validation, never in the plan.
diff --git a/src/runner/plugin/store/mod.rs b/src/runner/plugin/store/mod.rs
index e724fc8..c6d1438 100644
--- a/src/runner/plugin/store/mod.rs
+++ b/src/runner/plugin/store/mod.rs
@@ -37,7 +37,7 @@ use super::package::{
};
use super::platform::Platform;
use crate::error::ErrorChain;
-use crate::identifiers::{ExactVersion, ProgramName};
+use crate::identifiers::{ExactVersion, PluginProgramIdentity, ProgramName};
use crate::payload_contract::{
PluginInterface, PluginProgramPayloadContract, PluginProgramPayloadContractProjection,
};
@@ -52,8 +52,6 @@ use std::io::{self, Read};
use std::path::{Path, PathBuf};
use std::sync::Arc;
-type PluginProgramIdentity = (ProgramName, ExactVersion);
-
/// Installation disposition and the validated identity, without an execution lease.
#[derive(Debug)]
pub(crate) enum PluginProgramInstallResult {
@@ -128,8 +126,10 @@ impl PluginProgramStore {
program_name: &ProgramName,
exact_version: &ExactVersion,
) -> Option<&Arc<PluginProgramEntry>> {
- self.programs
- .get(&(program_name.clone(), exact_version.clone()))
+ self.programs.get(&PluginProgramIdentity::from_parts(
+ program_name.clone(),
+ exact_version.clone(),
+ ))
}
/// Iterates only validated Programs without consulting the filesystem.
@@ -138,7 +138,7 @@ impl PluginProgramStore {
) -> impl Iterator<Item = (&ProgramName, &ExactVersion, &Arc<PluginProgramEntry>)> {
self.programs
.iter()
- .map(|((program_name, exact_version), entry)| (program_name, exact_version, entry))
+ .map(|(identity, entry)| (identity.program_name(), identity.exact_version(), entry))
}
fn recover_with_filesystem(
@@ -174,7 +174,7 @@ impl PluginProgramStore {
match recovered {
Ok((exact_version, installed)) => {
store.programs.insert(
- (program_name.clone(), exact_version),
+ PluginProgramIdentity::from_parts(program_name.clone(), exact_version),
Arc::new(program_entry(target, installed)),
);
}
@@ -214,13 +214,17 @@ impl PluginProgramStore {
return Err(PluginStoreError::PlatformMismatch { platforms });
}
let program_name = staged.program_name().clone();
- let identity = (program_name.clone(), staged.exact_version().clone());
+ let identity =
+ PluginProgramIdentity::from_parts(program_name.clone(), staged.exact_version().clone());
let parent = self.directory.join(program_name.as_str());
let target = parent.join(staged.exact_version().as_str());
let namespace_presence = inspect_program_namespace(&parent)?;
if matches!(namespace_presence, ProgramNamespacePresence::Absent)
- && self.programs.keys().any(|(name, _)| name == &program_name)
+ && self
+ .programs
+ .keys()
+ .any(|identity| identity.program_name() == &program_name)
{
return Err(PluginStoreError::StoreIntegrityInvalid { path: parent });
}
@@ -271,7 +275,8 @@ impl PluginProgramStore {
exact_version: &ExactVersion,
publication: &impl PublicationFilesystem,
) -> Result<PluginUninstallOutcome, PluginStoreError> {
- let identity = (program_name.clone(), exact_version.clone());
+ let identity =
+ PluginProgramIdentity::from_parts(program_name.clone(), exact_version.clone());
let Some(entry) = self.programs.get_mut(&identity) else {
return Ok(PluginUninstallOutcome::NotFound);
};
diff --git a/src/runner/plugin/store/tests.rs b/src/runner/plugin/store/tests.rs
index 47553b3..51ab2b0 100644
--- a/src/runner/plugin/store/tests.rs
+++ b/src/runner/plugin/store/tests.rs
@@ -24,7 +24,7 @@ use super::{
PluginProgramEntry, PluginProgramStore, PluginStoreError, PluginUninstallOutcome,
PublicationFilesystem,
};
-use crate::identifiers::{ExactVersion, ProgramName};
+use crate::identifiers::{ExactVersion, PluginProgramIdentity, ProgramName};
use crate::payload_contract::PluginInterface;
use crate::runner::plugin::package::tests::STAGING_DIRECTORY_PREFIX;
use crate::runner::plugin::package::tests::{
@@ -50,7 +50,10 @@ fn first_install_publishes_disk_and_memory_in_one_store_mutation() -> io::Result
let PluginProgramInstallResult::Installed(identity) = outcome else {
return Err(io::Error::other("first install was reported as unchanged"));
};
- assert_eq!(identity, (source_program_name()?, exact_version()?));
+ assert_eq!(
+ identity,
+ PluginProgramIdentity::from_parts(source_program_name()?, exact_version()?)
+ );
let entry = available_entry(&store, "com.example.source")?;
assert_eq!(entry.directory(), root.join("com.example.source/1.0.0"));
@@ -106,7 +109,10 @@ fn equivalent_retry_reuses_the_same_entry_arc() -> io::Result<()> {
));
};
- assert_eq!(identity, (source_program_name()?, exact_version()?));
+ assert_eq!(
+ identity,
+ PluginProgramIdentity::from_parts(source_program_name()?, exact_version()?)
+ );
assert!(Arc::ptr_eq(
&first,
available_entry(&store, "com.example.source")?
@@ -895,7 +901,10 @@ fn java_target_bundles_install_idempotently_and_recover() -> io::Result<()> {
let PluginProgramInstallResult::Installed(identity) = outcome else {
return Err(io::Error::other("generated Java Program was not installed"));
};
- assert_eq!(identity, (program_name.clone(), version.clone()));
+ assert_eq!(
+ identity,
+ PluginProgramIdentity::from_parts(program_name.clone(), version.clone())
+ );
let entry = store
.lookup(&program_name, &version)
.cloned()
@@ -918,7 +927,10 @@ fn java_target_bundles_install_idempotently_and_recover() -> io::Result<()> {
"generated Java Program was installed twice",
));
};
- assert_eq!(identity, (program_name.clone(), version.clone()));
+ assert_eq!(
+ identity,
+ PluginProgramIdentity::from_parts(program_name.clone(), version.clone())
+ );
assert!(Arc::ptr_eq(
&entry,
store
diff --git a/src/tenon_document/verified/mod.rs b/src/tenon_document/verified/mod.rs
index c07aef8..33d40d6 100644
--- a/src/tenon_document/verified/mod.rs
+++ b/src/tenon_document/verified/mod.rs
@@ -34,7 +34,8 @@ use super::{SourceDelivery, UnverifiedTenonDocument};
use crate::config::ScriptVmLimits;
use crate::contracts::tenon_document::v1_schema_bytes;
use crate::identifiers::{
- ExactVersion, FlowId, PluginInstanceId, ProgramName, TenonDocumentId, validate_local_id,
+ ExactVersion, FlowId, PluginInstanceId, PluginProgramIdentity, ProgramName, TenonDocumentId,
+ validate_local_id,
};
use crate::lua::validate_tenon_document_source_syntax;
use crate::strict_jsonc::parse_json;
@@ -467,8 +468,8 @@ fn ceil_decimal_product(ratio: &Number, cpu_count: u128) -> u128 {
#[serde(rename_all = "camelCase")]
/// One verified use of a Plugin Program with its authored configuration.
pub struct PluginInstance {
- program_name: ProgramName,
- exact_version: ExactVersion,
+ #[serde(flatten)]
+ program_identity: PluginProgramIdentity,
#[serde(skip_serializing_if = "Option::is_none")]
extra_args: Option<ExtraArgs>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -477,14 +478,19 @@ pub struct PluginInstance {
}
impl PluginInstance {
+ /// Returns the exact Program identity shared by instances using this version.
+ pub fn program_identity(&self) -> &PluginProgramIdentity {
+ &self.program_identity
+ }
+
/// Returns the referenced Program name.
pub fn program_name(&self) -> &ProgramName {
- &self.program_name
+ self.program_identity.program_name()
}
/// Returns the exact referenced Program version.
pub fn exact_version(&self) -> &ExactVersion {
- &self.exact_version
+ self.program_identity.exact_version()
}
/// Returns the unchanged authored Plugin configuration.
@@ -547,8 +553,8 @@ impl fmt::Debug for PluginInstance {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("PluginInstance")
- .field("program_name", &self.program_name)
- .field("exact_version", &self.exact_version)
+ .field("program_name", self.program_identity.program_name())
+ .field("exact_version", self.program_identity.exact_version())
.field("config", &"[REDACTED]")
.finish()
}
@@ -643,13 +649,15 @@ impl RawDocument {
.into_iter()
.map(|(id, mut instance)| {
let instance = PluginInstance {
- program_name: ProgramName::from_verified(
- serde_json::from_value(instance["programName"].take())
- .expect("Schema guarantees a string Program name"),
- ),
- exact_version: ExactVersion::from_verified(
- serde_json::from_value(instance["exactVersion"].take())
- .expect("Schema guarantees a string exact version"),
+ program_identity: PluginProgramIdentity::from_parts(
+ ProgramName::from_verified(
+ serde_json::from_value(instance["programName"].take())
+ .expect("Schema guarantees a string Program name"),
+ ),
+ ExactVersion::from_verified(
+ serde_json::from_value(instance["exactVersion"].take())
+ .expect("Schema guarantees a string exact version"),
+ ),
),
extra_args: serde_json::from_value(instance["extraArgs"].take())
.expect("Schema guarantees optional Instance arguments"),