[PVM-host] Reject blocks with duplicate NEW service ids

pvm/pvm-host/src/context/mod.rs

@@ -188,6 +188,8 @@ pub struct AccumulateHostContext<S: HostStateProvider> {
     pub yielded_accumulate_hash: Option<AccumulationOutputHash>,
     /// **`p`**: Provided preimage data
     pub provided_preimages: HashSet<(ServiceId, Octets)>,
+    /// Service ids created during this accumulation
+    pub created_service_ids: HashSet<ServiceId>,
     /// Accumulate entry-point function invocation args (read-only)
     pub invoke_args: AccumulateInvokeArgs,
     /// Current entropy value (`η0′`)
@@ -203,6 +205,7 @@ impl<S: HostStateProvider> Clone for AccumulateHostContext<S> {
             deferred_transfers: self.deferred_transfers.clone(),
             yielded_accumulate_hash: self.yielded_accumulate_hash.clone(),
             provided_preimages: self.provided_preimages.clone(),
+            created_service_ids: self.created_service_ids.clone(),
             invoke_args: self.invoke_args.clone(),
             curr_entropy: self.curr_entropy.clone(),
         }
@@ -232,6 +235,7 @@ impl<S: HostStateProvider> AccumulateHostContext<S> {
             deferred_transfers: Vec::new(),
             yielded_accumulate_hash: None,
             provided_preimages: HashSet::new(),
+            created_service_ids: HashSet::new(),
             invoke_args,
             curr_entropy,
         })

pvm/pvm-host/src/context/partial_state.rs

@@ -278,7 +278,8 @@ impl<S: HostStateProvider> AccountsSandboxMap<S> {
 
     /// Returns `true` if the given service ID is already known, either in the accounts sandbox
     /// or in the global state. Even if the sandbox entry is marked for removed, returns `true`.
-    pub async fn account_exists_anywhere(
+    #[cfg(test)]
+    pub(crate) async fn account_exists_anywhere(
         &self,
         state_provider: Arc<S>,
         service_id: ServiceId,
@@ -290,6 +291,28 @@ impl<S: HostStateProvider> AccountsSandboxMap<S> {
         }
     }
 
+    /// Returns `true` if the account exists in the effective partial state:
+    /// - `Removed` entries are treated as non-existent.
+    /// - `Added`/`Updated`/`Clean` entries are treated as existing.
+    pub async fn account_exists_effective(
+        &self,
+        state_provider: Arc<S>,
+        service_id: ServiceId,
+    ) -> Result<bool, PartialStateError> {
+        if let Some(account_sandbox) = self.get(&service_id) {
+            if matches!(
+                account_sandbox.metadata.status(),
+                SandboxEntryStatus::Removed
+            ) {
+                Ok(false)
+            } else {
+                Ok(true)
+            }
+        } else {
+            Ok(state_provider.account_exists(service_id).await?)
+        }
+    }
+
     /// Returns `true` only when the sandbox still has a live copy of the account:
     /// the entry exists and hasn’t been marked for removal, or it was newly added but
     /// not yet added to the global state.
@@ -1303,7 +1326,7 @@ impl<S: HostStateProvider> AccumulatePartialState<S> {
         loop {
             if !self
                 .accounts_sandbox
-                .account_exists_anywhere(state_provider.clone(), check_id)
+                .account_exists_effective(state_provider.clone(), check_id)
                 .await?
             {
                 return Ok(check_id);

pvm/pvm-host/src/host_functions/accumulate/mod.rs

@@ -449,6 +449,7 @@ impl<S: HostStateProvider> AccumulateHostFunction<S> {
             x.rotate_new_account_id(state_provider).await?;
             new_service_id
         };
+        x.created_service_ids.insert(new_service_id);
 
         tracing::debug!(
             "NEW service_id={new_service_id} parent={}",

pvm/pvm-host/src/host_functions/general/tests.rs

@@ -297,6 +297,7 @@ mod fetch_tests {
                 deferred_transfers: vec![],
                 yielded_accumulate_hash: None,
                 provided_preimages: HashSet::new(),
+                created_service_ids: HashSet::new(),
                 invoke_args: AccumulateInvokeArgs {
                     inputs: self.accumulate_inputs.clone(),
                     ..Default::default()

pvm/pvm-invocation/src/accumulate/mod.rs

@@ -48,6 +48,8 @@ pub struct AccumulateResult<S: HostStateProvider> {
     pub gas_used: UnsignedGas,
     /// **`p`**: Provided preimage entries during accumulation
     pub provided_preimages: HashSet<(ServiceId, Octets)>,
+    /// Service ids created during accumulation
+    pub created_service_ids: HashSet<ServiceId>,
     pub accumulate_host: ServiceId,
 }
 
@@ -59,6 +61,7 @@ impl<S: HostStateProvider> Default for AccumulateResult<S> {
             yielded_accumulate_hash: None,
             gas_used: UnsignedGas::default(),
             provided_preimages: HashSet::new(),
+            created_service_ids: HashSet::new(),
             accumulate_host: ServiceId::default(),
         }
     }
@@ -236,6 +239,7 @@ impl<S: HostStateProvider> AccumulateInvocation<S> {
                     gas_used: result.gas_used,
                     accumulate_host: x.accumulate_host,
                     provided_preimages: x.provided_preimages,
+                    created_service_ids: x.created_service_ids,
                 }))
             }
             PVMInvocationOutput::OutputUnavailable => Ok(Some(AccumulateResult {
@@ -245,6 +249,7 @@ impl<S: HostStateProvider> AccumulateInvocation<S> {
                 gas_used: result.gas_used,
                 accumulate_host: x.accumulate_host,
                 provided_preimages: x.provided_preimages,
+                created_service_ids: x.created_service_ids,
             })),
             PVMInvocationOutput::OutOfGas(_) | PVMInvocationOutput::Panic(_) => {
                 Ok(Some(AccumulateResult {
@@ -254,6 +259,7 @@ impl<S: HostStateProvider> AccumulateInvocation<S> {
                     gas_used: result.gas_used, // Note: taking gas usage from the `x` context
                     accumulate_host: x.accumulate_host,
                     provided_preimages: y.provided_preimages,
+                    created_service_ids: y.created_service_ids,
                 }))
             }
         }

pvm/pvm-invocation/src/accumulate/pipeline.rs

@@ -188,6 +188,7 @@ async fn merge_partial_state_change(
     prev_assigns: &AssignServices,
     prev_designate: ServiceId,
     prev_registrar: ServiceId,
+    created_service_ids: &HashSet<ServiceId>,
 ) -> Result<(), PVMInvokeError> {
     // Accumulating service sandbox
     let accumulate_host_sandbox = partial_state_union
@@ -306,33 +307,42 @@ async fn merge_partial_state_change(
     // Integrate new accounts and ejected accounts.
     // Note: no account other than the accumulate host is ever touched except for the
     // `NEW` & `EJECT` hostcalls.
-    accumulate_result_partial_state
+    for (&service_id, sandbox) in accumulate_result_partial_state
         .accounts_sandbox
         .iter()
         .filter(|(&service_id, _)| service_id != accumulate_host)
-        .for_each(|(&service_id, sandbox)| {
-            match sandbox.metadata.status() {
-                SandboxEntryStatus::Added => {
-                    // Additional guard to avoid entries from the `partial_state_union` with `Added`
-                    // status being copied into the later accumulations and overwriting any updates.
-                    #[allow(clippy::map_entry)]
-                    if !partial_state_union
-                        .accounts_sandbox
-                        .contains_key(&service_id)
-                    {
+    {
+        match sandbox.metadata.status() {
+            SandboxEntryStatus::Added => {
+                match partial_state_union.accounts_sandbox.get(&service_id) {
+                    None => {
                         partial_state_union
                             .accounts_sandbox
                             .insert(service_id, sandbox.clone());
                     }
+                    Some(existing) => {
+                        if matches!(existing.metadata.status(), SandboxEntryStatus::Removed) {
+                            // Allow re-creating a previously removed account.
+                            partial_state_union
+                                .accounts_sandbox
+                                .insert(service_id, sandbox.clone());
+                        } else if created_service_ids.contains(&service_id) {
+                            // If NEW service ids collide, block should be considered invalid.
+                            return Err(PVMInvokeError::DuplicateNewServiceId(service_id));
+                        } else {
+                            // Inherited Added entry from a prior round; skip to avoid overwriting.
+                        }
+                    }
                 }
-                SandboxEntryStatus::Removed => {
-                    partial_state_union
-                        .accounts_sandbox
-                        .insert(service_id, sandbox.clone());
-                }
-                _ => {}
             }
-        });
+            SandboxEntryStatus::Removed => {
+                partial_state_union
+                    .accounts_sandbox
+                    .insert(service_id, sandbox.clone());
+            }
+            _ => {}
+        }
+    }
 
     Ok(())
 }
@@ -487,6 +497,7 @@ async fn accumulate_parallel(
                 &prev_assigns,
                 prev_designate,
                 prev_registrar,
+                &accumulate_result.created_service_ids,
             )
             .await?;
 

pvm/pvm-invocation/src/error.rs

@@ -23,6 +23,8 @@ pub enum PVMInvokeError {
     WorkReportBlobTooLarge,
     #[error("Account sandbox for accumulate host (s={0}) is missing")]
     MissingAccumulateHostSandbox(ServiceId),
+    #[error("Duplicate new service id (s={0})")]
+    DuplicateNewServiceId(ServiceId),
     #[error("Number of work digests exceeds maximum")]
     WorkDigestsOverflow,
     #[error("Gas usage or gas limit overflowed")]