@@ -17,8 +17,8 @@ use inferadb_ledger_state::{
1717} ;
1818use inferadb_ledger_store:: StorageBackend ;
1919use inferadb_ledger_types:: {
20- AppId , AppSlug , Hash , LedgerErrorCode , Operation , OrganizationId , TeamId , TeamSlug ,
21- TokenSubject , TokenType , VaultEntry , VaultId , compute_tx_merkle_root, decode, encode,
20+ AppId , AppSlug , Hash , LedgerErrorCode , Operation , OrganizationId , SetCondition , TeamId ,
21+ TeamSlug , TokenSubject , TokenType , VaultEntry , VaultId , compute_tx_merkle_root, decode, encode,
2222 events:: { EventAction , EventEntry , EventOutcome } ,
2323} ;
2424
@@ -58,33 +58,55 @@ fn try_encode<T: serde::Serialize>(value: &T, context: &str) -> Option<Vec<u8>>
5858/// the record on its next poll cycle.
5959///
6060/// This is a synchronous write through the state layer (not Raft) because
61- /// the apply handler is already inside a Raft commit.
61+ /// the apply handler is already inside a Raft commit. The saga ID is derived
62+ /// deterministically from the scope to ensure all replicas produce identical
63+ /// state when applying the same log entry.
6264fn write_signing_key_saga < B : StorageBackend > (
6365 state_layer : & Arc < StateLayer < B > > ,
6466 scope : SigningKeyScope ,
6567) {
66- let saga_id = uuid:: Uuid :: new_v4 ( ) . to_string ( ) ;
68+ let saga_id = match & scope {
69+ SigningKeyScope :: Global => "create-signing-key-global" . to_owned ( ) ,
70+ SigningKeyScope :: Organization ( org_id) => {
71+ format ! ( "create-signing-key-org-{}" , org_id. value( ) )
72+ } ,
73+ } ;
6774 let saga = CreateSigningKeySaga :: new ( saga_id. clone ( ) , CreateSigningKeyInput { scope } ) ;
6875 let wrapped = Saga :: CreateSigningKey ( saga) ;
6976
7077 let key = format ! ( "saga:{saga_id}" ) ;
7178 match serde_json:: to_vec ( & wrapped) {
7279 Ok ( value) => {
73- let op =
74- Operation :: SetEntity { key : key. clone ( ) , value, condition : None , expires_at : None } ;
75- if let Err ( e) = state_layer. apply_operations ( SYSTEM_VAULT_ID , & [ op] , 0 ) {
76- tracing:: error!(
77- saga_id = %saga_id,
78- scope = ?scope,
79- error = %e,
80- "Failed to write CreateSigningKeySaga record"
81- ) ;
82- } else {
83- tracing:: info!(
84- saga_id = %saga_id,
85- scope = ?scope,
86- "Wrote CreateSigningKeySaga from apply handler"
87- ) ;
80+ let op = Operation :: SetEntity {
81+ key : key. clone ( ) ,
82+ value,
83+ condition : Some ( SetCondition :: MustNotExist ) ,
84+ expires_at : None ,
85+ } ;
86+ match state_layer. apply_operations ( SYSTEM_VAULT_ID , & [ op] , 0 ) {
87+ Ok ( _) => {
88+ tracing:: info!(
89+ saga_id = %saga_id,
90+ scope = ?scope,
91+ "Wrote CreateSigningKeySaga from apply handler"
92+ ) ;
93+ } ,
94+ Err ( StateError :: PreconditionFailed { .. } ) => {
95+ // Expected on log replay — saga already exists from prior apply.
96+ tracing:: info!(
97+ saga_id = %saga_id,
98+ scope = ?scope,
99+ "CreateSigningKeySaga already exists (idempotent replay)"
100+ ) ;
101+ } ,
102+ Err ( e) => {
103+ tracing:: error!(
104+ saga_id = %saga_id,
105+ scope = ?scope,
106+ error = %e,
107+ "Failed to write CreateSigningKeySaga record"
108+ ) ;
109+ } ,
88110 }
89111 } ,
90112 Err ( e) => {
@@ -4138,41 +4160,6 @@ impl<B: StorageBackend> RaftLogStore<B> {
41384160 }
41394161 } ,
41404162
4141- LedgerRequest :: RevokeAllSubjectTokens { subject } => {
4142- let Some ( state_layer) = & self . state_layer else {
4143- return error_result ( LedgerErrorCode :: Internal , "State layer not available" ) ;
4144- } ;
4145- let sys = SystemOrganizationService :: new ( state_layer. clone ( ) ) ;
4146-
4147- match sys. revoke_all_subject_tokens ( subject, block_timestamp) {
4148- Ok ( result) => {
4149- ( LedgerResponse :: SubjectTokensRevoked { count : result. revoked_count } , None )
4150- } ,
4151- Err ( e) => error_result (
4152- LedgerErrorCode :: Internal ,
4153- format ! ( "Failed to revoke subject tokens: {e}" ) ,
4154- ) ,
4155- }
4156- } ,
4157-
4158- LedgerRequest :: RevokeAppVaultTokens { app, vault } => {
4159- let Some ( state_layer) = & self . state_layer else {
4160- return error_result ( LedgerErrorCode :: Internal , "State layer not available" ) ;
4161- } ;
4162- let sys = SystemOrganizationService :: new ( state_layer. clone ( ) ) ;
4163-
4164- match sys. revoke_app_vault_tokens ( * app, * vault, block_timestamp) {
4165- Ok ( result) => (
4166- LedgerResponse :: AppVaultTokensRevoked { count : result. revoked_count } ,
4167- None ,
4168- ) ,
4169- Err ( e) => error_result (
4170- LedgerErrorCode :: Internal ,
4171- format ! ( "Failed to revoke app vault tokens: {e}" ) ,
4172- ) ,
4173- }
4174- } ,
4175-
41764163 LedgerRequest :: RevokeAllUserSessions { user } => {
41774164 let Some ( state_layer) = & self . state_layer else {
41784165 return error_result ( LedgerErrorCode :: Internal , "State layer not available" ) ;
0 commit comments