diff --git a/pkg/protocol/engine/booker/inmemorybooker/booker.go b/pkg/protocol/engine/booker/inmemorybooker/booker.go
index f84a226f2..8b360baf6 100644
--- a/pkg/protocol/engine/booker/inmemorybooker/booker.go
+++ b/pkg/protocol/engine/booker/inmemorybooker/booker.go
@@ -114,6 +114,13 @@ func (b *Booker) Queue(block *blocks.Block) error {
 	// Based on the assumption that we always fork and the UTXO and Tangle past cones are always fully known.
 	signedTransactionMetadata.OnSignaturesValid(func() {
 		transactionMetadata := signedTransactionMetadata.TransactionMetadata()
+
+		if orphanedSlot, isOrphaned := transactionMetadata.OrphanedSlot(); isOrphaned && orphanedSlot <= block.SlotCommitmentID().Slot() {
+			block.SetInvalid()
+
+			return
+		}
+
 		transactionMetadata.OnBooked(func() {
 			block.SetPayloadConflictIDs(transactionMetadata.ConflictIDs())
 			b.bookingOrder.Queue(block)
@@ -142,6 +149,21 @@ func (b *Booker) book(block *blocks.Block) error {
 		return ierrors.Wrapf(err, "failed to inherit conflicts for block %s", block.ID())
 	}
 
+	// The block is invalid if it carries a conflict that has been orphaned with respect to its commitment.
+	for it := conflictsToInherit.Iterator(); it.HasNext(); {
+		conflictID := it.Next()
+
+		txMetadata, exists := b.ledger.MemPool().TransactionMetadata(conflictID)
+		if !exists {
+			return ierrors.Errorf("failed to load transaction %s for block %s", conflictID.String(), block.ID())
+		}
+
+		if orphanedSlot, orphaned := txMetadata.OrphanedSlot(); orphaned && orphanedSlot <= block.SlotCommitmentID().Slot() {
+			// Merge-to-master orphaned conflicts.
+			conflictsToInherit.Delete(conflictID)
+		}
+	}
+
 	block.SetConflictIDs(conflictsToInherit)
 	block.SetBooked()
 	b.events.BlockBooked.Trigger(block)
diff --git a/pkg/protocol/engine/ledger/ledger/ledger.go b/pkg/protocol/engine/ledger/ledger/ledger.go
index fd546ae77..bddd760d4 100644
--- a/pkg/protocol/engine/ledger/ledger/ledger.go
+++ b/pkg/protocol/engine/ledger/ledger/ledger.go
@@ -68,7 +68,7 @@ func NewProvider() module.Provider[*engine.Engine, ledger.Ledger] {
 
 			l.setRetainTransactionFailureFunc(e.Retainer.RetainTransactionFailure)
 
-			l.memPool = mempoolv1.New(NewVM(l), l.resolveState, e.Workers.CreateGroup("MemPool"), l.conflictDAG, l.errorHandler, mempoolv1.WithForkAllTransactions[ledger.BlockVoteRank](true))
+			l.memPool = mempoolv1.New(NewVM(l), l.resolveState, e.Workers.CreateGroup("MemPool"), l.conflictDAG, l.apiProvider, l.errorHandler, mempoolv1.WithForkAllTransactions[ledger.BlockVoteRank](true))
 			e.EvictionState.Events.SlotEvicted.Hook(l.memPool.Evict)
 
 			l.manaManager = mana.NewManager(l.apiProvider, l.resolveAccountOutput, l.accountsLedger.Account)
diff --git a/pkg/protocol/engine/mempool/tests/tests.go b/pkg/protocol/engine/mempool/tests/tests.go
index 33cb00f0b..c14fdd591 100644
--- a/pkg/protocol/engine/mempool/tests/tests.go
+++ b/pkg/protocol/engine/mempool/tests/tests.go
@@ -11,6 +11,7 @@ import (
 	"github.com/iotaledger/hive.go/runtime/memanalyzer"
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/mempool"
 	iotago "github.com/iotaledger/iota.go/v4"
+	"github.com/iotaledger/iota.go/v4/tpkg"
 )
 
 const (
@@ -210,11 +211,10 @@ func TestSetInclusionSlot(t *testing.T, tf *TestFramework) {
 	require.True(t, exists)
 
 	tf.CommitSlot(1)
-	// time.Sleep(1 * time.Second)
-	transactionDeletionState := map[string]bool{"tx1": true, "tx2": false, "tx3": false}
+	transactionDeletionState := map[string]bool{"tx1": false, "tx2": false, "tx3": false}
 	tf.RequireTransactionsEvicted(transactionDeletionState)
 
-	attachmentDeletionState := map[string]bool{"block1": true, "block2": false, "block3": false}
+	attachmentDeletionState := map[string]bool{"block1": false, "block2": false, "block3": false}
 	tf.RequireAttachmentsEvicted(attachmentDeletionState)
 
 	tf.Instance.Evict(1)
@@ -223,9 +223,8 @@ func TestSetInclusionSlot(t *testing.T, tf *TestFramework) {
 	tf.RequireBooked("tx3")
 
 	tf.CommitSlot(2)
-	// time.Sleep(1 * time.Second)
-	tf.RequireTransactionsEvicted(lo.MergeMaps(transactionDeletionState, map[string]bool{"tx2": true}))
-	tf.RequireAttachmentsEvicted(lo.MergeMaps(attachmentDeletionState, map[string]bool{"block2": true}))
+	tf.RequireTransactionsEvicted(lo.MergeMaps(transactionDeletionState, map[string]bool{"tx2": false}))
+	tf.RequireAttachmentsEvicted(lo.MergeMaps(attachmentDeletionState, map[string]bool{"block1": true}))
 
 	tf.Instance.Evict(2)
 	tf.RequireBooked("tx3")
@@ -234,14 +233,13 @@ func TestSetInclusionSlot(t *testing.T, tf *TestFramework) {
 	tf.RequireAccepted(map[string]bool{"tx3": true})
 
 	tf.CommitSlot(3)
-	// time.Sleep(1 * time.Second)
-	tf.RequireTransactionsEvicted(lo.MergeMaps(transactionDeletionState, map[string]bool{"tx3": true}))
+	tf.RequireTransactionsEvicted(transactionDeletionState)
 
-	require.False(t, tx1Metadata.IsOrphaned())
-	require.False(t, tx2Metadata.IsOrphaned())
-	require.False(t, tx3Metadata.IsOrphaned())
+	require.False(t, lo.Return2(tx1Metadata.OrphanedSlot()))
+	require.False(t, lo.Return2(tx2Metadata.OrphanedSlot()))
+	require.False(t, lo.Return2(tx3Metadata.OrphanedSlot()))
 
-	tf.RequireAttachmentsEvicted(lo.MergeMaps(attachmentDeletionState, map[string]bool{"block3": true}))
+	tf.RequireAttachmentsEvicted(lo.MergeMaps(attachmentDeletionState, map[string]bool{"block1": true, "block2": true, "block3": false}))
 }
 
 func TestSetTransactionOrphanage(t *testing.T, tf *TestFramework) {
@@ -273,13 +271,14 @@ func TestSetTransactionOrphanage(t *testing.T, tf *TestFramework) {
 
 	tf.Instance.Evict(1)
 
-	tf.RequireTransactionsEvicted(map[string]bool{"tx1": true, "tx2": true, "tx3": true})
+	// We only evict after MCA
+	tf.RequireTransactionsEvicted(map[string]bool{"tx1": false, "tx2": false, "tx3": false})
 
-	require.True(t, tx1Metadata.IsOrphaned())
-	require.True(t, tx2Metadata.IsOrphaned())
-	require.True(t, tx3Metadata.IsOrphaned())
+	require.True(t, lo.Return2(tx1Metadata.OrphanedSlot()))
+	require.True(t, tx2Metadata.IsPending())
+	require.True(t, tx3Metadata.IsPending())
 
-	tf.RequireAttachmentsEvicted(map[string]bool{"block1": true, "block2": true, "block3": true})
+	tf.RequireAttachmentsEvicted(map[string]bool{"block1": true, "block2": false, "block3": false})
 }
 
 func TestSetTxOrphanageMultipleAttachments(t *testing.T, tf *TestFramework) {
@@ -313,9 +312,9 @@ func TestSetTxOrphanageMultipleAttachments(t *testing.T, tf *TestFramework) {
 	require.False(t, tx3Metadata.IsAccepted())
 
 	tf.Instance.Evict(1)
-	require.False(t, tx1Metadata.IsOrphaned())
-	require.False(t, tx2Metadata.IsOrphaned())
-	require.False(t, tx3Metadata.IsOrphaned())
+	require.False(t, lo.Return2(tx1Metadata.OrphanedSlot()))
+	require.False(t, lo.Return2(tx2Metadata.OrphanedSlot()))
+	require.False(t, lo.Return2(tx3Metadata.OrphanedSlot()))
 
 	require.True(t, lo.Return2(tf.ConflictDAG.ConflictSets(tf.TransactionID("tx1"))))
 	require.True(t, lo.Return2(tf.ConflictDAG.ConflictSets(tf.TransactionID("tx2"))))
@@ -323,17 +322,18 @@ func TestSetTxOrphanageMultipleAttachments(t *testing.T, tf *TestFramework) {
 
 	tf.Instance.Evict(2)
 
-	require.True(t, tx1Metadata.IsOrphaned())
-	require.True(t, tx2Metadata.IsOrphaned())
-	require.True(t, tx3Metadata.IsOrphaned())
+	require.True(t, lo.Return2(tx1Metadata.OrphanedSlot()))
+	require.True(t, lo.Return2(tx2Metadata.OrphanedSlot()))
+	require.True(t, lo.Return2(tx3Metadata.OrphanedSlot()))
 
-	require.False(t, lo.Return2(tf.ConflictDAG.ConflictSets(tf.TransactionID("tx1"))))
-	require.False(t, lo.Return2(tf.ConflictDAG.ConflictSets(tf.TransactionID("tx2"))))
-	require.False(t, lo.Return2(tf.ConflictDAG.ConflictSets(tf.TransactionID("tx3"))))
+	// All conflicts still exist, as they are kept around until MCA
+	require.True(t, lo.Return2(tf.ConflictDAG.ConflictSets(tf.TransactionID("tx1"))))
+	require.True(t, lo.Return2(tf.ConflictDAG.ConflictSets(tf.TransactionID("tx2"))))
+	require.True(t, lo.Return2(tf.ConflictDAG.ConflictSets(tf.TransactionID("tx3"))))
 
-	tf.RequireTransactionsEvicted(map[string]bool{"tx1": true, "tx2": true, "tx3": true})
+	tf.RequireTransactionsEvicted(map[string]bool{"tx1": false, "tx2": false, "tx3": false})
 
-	tf.RequireAttachmentsEvicted(map[string]bool{"block1.1": true, "block1.2": true, "block2": true, "block3": true})
+	tf.RequireAttachmentsEvicted(map[string]bool{"block1.1": true, "block1.2": true, "block2": false, "block3": false})
 }
 
 func TestStateDiff(t *testing.T, tf *TestFramework) {
@@ -482,7 +482,13 @@ func TestMemoryRelease(t *testing.T, tf *TestFramework) {
 	txIndex, prevStateAlias := issueTransactions(1, 20000, "genesis")
 	tf.WaitChildren()
 
-	issueTransactions(txIndex, 20000, prevStateAlias)
+	txIndex, _ = issueTransactions(txIndex, 20000, prevStateAlias)
+
+	// Eviction is delayed by MCA, so we force Commit and Eviction.
+	for index := txIndex; index <= txIndex+int(tpkg.TestAPI.ProtocolParameters().MaxCommittableAge()); index++ {
+		tf.CommitSlot(iotago.SlotIndex(index))
+		tf.Instance.Evict(iotago.SlotIndex(index))
+	}
 
 	tf.Cleanup()
 
diff --git a/pkg/protocol/engine/mempool/transaction_metadata.go b/pkg/protocol/engine/mempool/transaction_metadata.go
index 3b1103ad8..e594a7d5a 100644
--- a/pkg/protocol/engine/mempool/transaction_metadata.go
+++ b/pkg/protocol/engine/mempool/transaction_metadata.go
@@ -57,15 +57,15 @@ type inclusionFlags interface {
 
 	OnAccepted(callback func())
 
-	IsCommitted() bool
+	CommittedSlot() (slot iotago.SlotIndex, isCommitted bool)
 
-	OnCommitted(callback func())
+	OnCommittedSlotUpdated(callback func(slot iotago.SlotIndex))
 
 	IsRejected() bool
 
 	OnRejected(callback func())
 
-	IsOrphaned() bool
+	OrphanedSlot() (slot iotago.SlotIndex, isOrphaned bool)
 
-	OnOrphaned(callback func())
+	OnOrphanedSlotUpdated(callback func(slot iotago.SlotIndex))
 }
diff --git a/pkg/protocol/engine/mempool/v1/inclusion_flags.go b/pkg/protocol/engine/mempool/v1/inclusion_flags.go
index 7774fddb8..3b8213690 100644
--- a/pkg/protocol/engine/mempool/v1/inclusion_flags.go
+++ b/pkg/protocol/engine/mempool/v1/inclusion_flags.go
@@ -3,6 +3,7 @@ package mempoolv1
 import (
 	"github.com/iotaledger/hive.go/ds/reactive"
 	"github.com/iotaledger/hive.go/runtime/promise"
+	iotago "github.com/iotaledger/iota.go/v4"
 )
 
 // inclusionFlags represents important flags and events that relate to the inclusion of an entity in the distributed ledger.
@@ -10,23 +11,30 @@ type inclusionFlags struct {
 	// accepted gets triggered when the entity gets marked as accepted.
 	accepted reactive.Variable[bool]
 
-	// committed gets triggered when the entity gets marked as committed.
-	committed *promise.Event
+	// committedSlot gets set to the slot in which the entity gets marked as committed.
+	committedSlot reactive.Variable[iotago.SlotIndex]
 
 	// rejected gets triggered when the entity gets marked as rejected.
 	rejected *promise.Event
 
-	// orphaned gets triggered when the entity gets marked as orphaned.
-	orphaned *promise.Event
+	// orphanedSlot gets set to the slot in which the entity gets marked as orphaned.
+	orphanedSlot reactive.Variable[iotago.SlotIndex]
 }
 
 // newInclusionFlags creates a new inclusionFlags instance.
 func newInclusionFlags() *inclusionFlags {
 	return &inclusionFlags{
-		accepted:  reactive.NewVariable[bool](),
-		committed: promise.NewEvent(),
-		rejected:  promise.NewEvent(),
-		orphaned:  promise.NewEvent(),
+		accepted:      reactive.NewVariable[bool](),
+		committedSlot: reactive.NewVariable[iotago.SlotIndex](),
+		rejected:      promise.NewEvent(),
+		// Make sure the oldest orphaned index doesn't get overridden by newer TX spending the orphaned conflict further.
+		orphanedSlot: reactive.NewVariable[iotago.SlotIndex](func(currentValue, newValue iotago.SlotIndex) iotago.SlotIndex {
+			if currentValue != 0 {
+				return currentValue
+			}
+
+			return newValue
+		}),
 	}
 }
 
@@ -68,21 +76,25 @@ func (s *inclusionFlags) OnRejected(callback func()) {
 }
 
 // IsCommitted returns true if the entity was committed.
-func (s *inclusionFlags) IsCommitted() bool {
-	return s.committed.WasTriggered()
+func (s *inclusionFlags) CommittedSlot() (slot iotago.SlotIndex, isCommitted bool) {
+	return s.committedSlot.Get(), s.committedSlot.Get() != 0
 }
 
 // OnCommitted registers a callback that gets triggered when the entity gets committed.
-func (s *inclusionFlags) OnCommitted(callback func()) {
-	s.committed.OnTrigger(callback)
+func (s *inclusionFlags) OnCommittedSlotUpdated(callback func(slot iotago.SlotIndex)) {
+	s.committedSlot.OnUpdate(func(_, newValue iotago.SlotIndex) {
+		callback(newValue)
+	})
 }
 
 // IsOrphaned returns true if the entity was orphaned.
-func (s *inclusionFlags) IsOrphaned() bool {
-	return s.orphaned.WasTriggered()
+func (s *inclusionFlags) OrphanedSlot() (slot iotago.SlotIndex, isOrphaned bool) {
+	return s.orphanedSlot.Get(), s.orphanedSlot.Get() != 0
 }
 
 // OnOrphaned registers a callback that gets triggered when the entity gets orphaned.
-func (s *inclusionFlags) OnOrphaned(callback func()) {
-	s.orphaned.OnTrigger(callback)
+func (s *inclusionFlags) OnOrphanedSlotUpdated(callback func(slot iotago.SlotIndex)) {
+	s.orphanedSlot.OnUpdate(func(_, newValue iotago.SlotIndex) {
+		callback(newValue)
+	})
 }
diff --git a/pkg/protocol/engine/mempool/v1/mempool.go b/pkg/protocol/engine/mempool/v1/mempool.go
index 8e3c80a60..4f1905cd1 100644
--- a/pkg/protocol/engine/mempool/v1/mempool.go
+++ b/pkg/protocol/engine/mempool/v1/mempool.go
@@ -42,9 +42,19 @@ type MemPool[VoteRank conflictdag.VoteRankType[VoteRank]] struct {
 	// stateDiffs holds aggregated state mutations for each slot.
 	stateDiffs *shrinkingmap.ShrinkingMap[iotago.SlotIndex, *StateDiff]
 
+	// delayedTransactionEviction holds the transactions that can only be evicted after MaxCommittableAge to objectively
+	// invalidate blocks that try to spend from them.
+	delayedTransactionEviction *shrinkingmap.ShrinkingMap[iotago.SlotIndex, ds.Set[iotago.TransactionID]]
+
+	// delayedOutputStateEviction holds the outputs that can only be evicted after MaxCommittableAge to objectively
+	// invalidate blocks that try to spend them.
+	delayedOutputStateEviction *shrinkingmap.ShrinkingMap[iotago.SlotIndex, *shrinkingmap.ShrinkingMap[iotago.Identifier, *StateMetadata]]
+
 	// conflictDAG is the DAG that is used to keep track of the conflicts between transactions.
 	conflictDAG conflictdag.ConflictDAG[iotago.TransactionID, mempool.StateID, VoteRank]
 
+	apiProvider iotago.APIProvider
+
 	errorHandler func(error)
 
 	// executionWorkers is the worker pool that is used to execute the state transitions of transactions.
@@ -69,22 +79,26 @@ func New[VoteRank conflictdag.VoteRankType[VoteRank]](
 	stateResolver mempool.StateResolver,
 	workers *workerpool.Group,
 	conflictDAG conflictdag.ConflictDAG[iotago.TransactionID, mempool.StateID, VoteRank],
+	apiProvider iotago.APIProvider,
 	errorHandler func(error),
 	opts ...options.Option[MemPool[VoteRank]],
 ) *MemPool[VoteRank] {
 	return options.Apply(&MemPool[VoteRank]{
-		vm:                        vm,
-		resolveState:              stateResolver,
-		attachments:               memstorage.NewIndexedStorage[iotago.SlotIndex, iotago.BlockID, *SignedTransactionMetadata](),
-		cachedTransactions:        shrinkingmap.New[iotago.TransactionID, *TransactionMetadata](),
-		cachedSignedTransactions:  shrinkingmap.New[iotago.SignedTransactionID, *SignedTransactionMetadata](),
-		cachedStateRequests:       shrinkingmap.New[mempool.StateID, *promise.Promise[*StateMetadata]](),
-		stateDiffs:                shrinkingmap.New[iotago.SlotIndex, *StateDiff](),
-		executionWorkers:          workers.CreatePool("executionWorkers", 1),
-		conflictDAG:               conflictDAG,
-		errorHandler:              errorHandler,
-		signedTransactionAttached: event.New1[mempool.SignedTransactionMetadata](),
-		transactionAttached:       event.New1[mempool.TransactionMetadata](),
+		vm:                         vm,
+		resolveState:               stateResolver,
+		attachments:                memstorage.NewIndexedStorage[iotago.SlotIndex, iotago.BlockID, *SignedTransactionMetadata](),
+		cachedTransactions:         shrinkingmap.New[iotago.TransactionID, *TransactionMetadata](),
+		cachedSignedTransactions:   shrinkingmap.New[iotago.SignedTransactionID, *SignedTransactionMetadata](),
+		cachedStateRequests:        shrinkingmap.New[mempool.StateID, *promise.Promise[*StateMetadata]](),
+		stateDiffs:                 shrinkingmap.New[iotago.SlotIndex, *StateDiff](),
+		executionWorkers:           workers.CreatePool("executionWorkers", 1),
+		delayedTransactionEviction: shrinkingmap.New[iotago.SlotIndex, ds.Set[iotago.TransactionID]](),
+		delayedOutputStateEviction: shrinkingmap.New[iotago.SlotIndex, *shrinkingmap.ShrinkingMap[iotago.Identifier, *StateMetadata]](),
+		conflictDAG:                conflictDAG,
+		apiProvider:                apiProvider,
+		errorHandler:               errorHandler,
+		signedTransactionAttached:  event.New1[mempool.SignedTransactionMetadata](),
+		transactionAttached:        event.New1[mempool.TransactionMetadata](),
 	}, opts, (*MemPool[VoteRank]).setup)
 }
 
@@ -183,6 +197,30 @@ func (m *MemPool[VoteRank]) Evict(slot iotago.SlotIndex) {
 			return true
 		})
 	}
+
+	maxCommittableAge := m.apiProvider.APIForSlot(slot).ProtocolParameters().MaxCommittableAge()
+	if slot <= maxCommittableAge {
+		return
+	}
+
+	delayedEvictionSlot := slot - maxCommittableAge
+	if delayedTransactions, exists := m.delayedTransactionEviction.Get(delayedEvictionSlot); exists {
+		delayedTransactions.Range(func(txID iotago.TransactionID) {
+			if transaction, exists := m.cachedTransactions.Get(txID); exists {
+				transaction.setEvicted()
+			}
+		})
+		m.delayedTransactionEviction.Delete(delayedEvictionSlot)
+	}
+
+	if delayedOutputs, exists := m.delayedOutputStateEviction.Get(delayedEvictionSlot); exists {
+		delayedOutputs.ForEach(func(stateID iotago.Identifier, state *StateMetadata) bool {
+			m.cachedStateRequests.Delete(stateID, state.HasNoSpenders)
+
+			return true
+		})
+		m.delayedOutputStateEviction.Delete(delayedEvictionSlot)
+	}
 }
 
 func (m *MemPool[VoteRank]) storeTransaction(signedTransaction mempool.SignedTransaction, transaction mempool.Transaction, blockID iotago.BlockID) (storedSignedTransaction *SignedTransactionMetadata, isNewSignedTransaction, isNewTransaction bool, err error) {
@@ -282,7 +320,8 @@ func (m *MemPool[VoteRank]) bookTransaction(transaction *TransactionMetadata) {
 		})
 	}
 
-	if !transaction.IsOrphaned() && transaction.setBooked() {
+	// if !lo.Return2(transaction.IsOrphaned()) && transaction.setBooked() {
+	if transaction.setBooked() {
 		m.publishOutputStates(transaction)
 	}
 }
@@ -291,7 +330,8 @@ func (m *MemPool[VoteRank]) forkTransaction(transactionMetadata *TransactionMeta
 	transactionMetadata.conflicting.Trigger()
 
 	if err := m.conflictDAG.UpdateConflictingResources(transactionMetadata.ID(), resourceIDs); err != nil {
-		transactionMetadata.orphaned.Trigger()
+		// this is a hack, as with a reactive.Variable we cannot set it to 0 and still check if it was orphaned.
+		transactionMetadata.orphanedSlot.Set(1)
 
 		m.errorHandler(err)
 	}
@@ -315,11 +355,10 @@ func (m *MemPool[VoteRank]) requestState(stateRef mempool.StateReference, waitIf
 		request.OnSuccess(func(state mempool.State) {
 			// The output was resolved from the ledger, meaning it was actually persisted as it was accepted and
 			// committed: otherwise we would have found it in cache or the request would have never resolved.
-			outputStateMetadata := NewStateMetadata(state)
-			outputStateMetadata.accepted.Set(true)
-			outputStateMetadata.committed.Trigger()
+			stateMetadata := NewStateMetadata(state)
+			stateMetadata.accepted.Set(true)
 
-			p.Resolve(outputStateMetadata)
+			p.Resolve(stateMetadata)
 		})
 
 		request.OnError(func(err error) {
@@ -440,17 +479,31 @@ func (m *MemPool[VoteRank]) setupTransaction(transaction *TransactionMetadata) {
 
 		transaction.signingTransactions.Range((*SignedTransactionMetadata).setEvicted)
 	})
+
+	transaction.OnCommittedSlotUpdated(func(slot iotago.SlotIndex) {
+		lo.Return1(m.delayedTransactionEviction.GetOrCreate(slot, func() ds.Set[iotago.TransactionID] { return ds.NewSet[iotago.TransactionID]() })).Add(transaction.ID())
+	})
+
+	transaction.OnOrphanedSlotUpdated(func(slot iotago.SlotIndex) {
+		lo.Return1(m.delayedTransactionEviction.GetOrCreate(slot, func() ds.Set[iotago.TransactionID] { return ds.NewSet[iotago.TransactionID]() })).Add(transaction.ID())
+	})
 }
 
 func (m *MemPool[VoteRank]) setupOutputState(stateMetadata *StateMetadata) {
-	stateMetadata.OnCommitted(func() {
-		if !m.cachedStateRequests.Delete(stateMetadata.state.StateID(), stateMetadata.HasNoSpenders) && m.cachedStateRequests.Has(stateMetadata.state.StateID()) {
-			stateMetadata.onAllSpendersRemoved(func() { m.cachedStateRequests.Delete(stateMetadata.state.StateID(), stateMetadata.HasNoSpenders) })
-		}
+	stateMetadata.onAllSpendersRemoved(func() {
+		m.cachedStateRequests.Delete(stateMetadata.state.StateID(), stateMetadata.HasNoSpenders)
+	})
+
+	stateMetadata.OnCommittedSlotUpdated(func(slot iotago.SlotIndex) {
+		lo.Return1(m.delayedOutputStateEviction.GetOrCreate(slot, func() *shrinkingmap.ShrinkingMap[iotago.Identifier, *StateMetadata] {
+			return shrinkingmap.New[iotago.Identifier, *StateMetadata]()
+		})).Set(stateMetadata.state.StateID(), stateMetadata)
 	})
 
-	stateMetadata.OnOrphaned(func() {
-		m.cachedStateRequests.Delete(stateMetadata.state.StateID())
+	stateMetadata.OnOrphanedSlotUpdated(func(slot iotago.SlotIndex) {
+		lo.Return1(m.delayedOutputStateEviction.GetOrCreate(slot, func() *shrinkingmap.ShrinkingMap[iotago.Identifier, *StateMetadata] {
+			return shrinkingmap.New[iotago.Identifier, *StateMetadata]()
+		})).Set(stateMetadata.state.StateID(), stateMetadata)
 	})
 }
 
diff --git a/pkg/protocol/engine/mempool/v1/mempool_test.go b/pkg/protocol/engine/mempool/v1/mempool_test.go
index c62e5339a..25aa9746c 100644
--- a/pkg/protocol/engine/mempool/v1/mempool_test.go
+++ b/pkg/protocol/engine/mempool/v1/mempool_test.go
@@ -19,6 +19,8 @@ import (
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/mempool/conflictdag/conflictdagv1"
 	mempooltests "github.com/iotaledger/iota-core/pkg/protocol/engine/mempool/tests"
 	iotago "github.com/iotaledger/iota.go/v4"
+	"github.com/iotaledger/iota.go/v4/api"
+	"github.com/iotaledger/iota.go/v4/tpkg"
 )
 
 func TestMemPoolV1_InterfaceWithoutForkingEverything(t *testing.T) {
@@ -36,7 +38,7 @@ func TestMempoolV1_ResourceCleanup(t *testing.T) {
 	conflictDAG := conflictdagv1.New[iotago.TransactionID, mempool.StateID, vote.MockedRank](func() int { return 0 })
 	memPoolInstance := New[vote.MockedRank](new(mempooltests.VM), func(reference mempool.StateReference) *promise.Promise[mempool.State] {
 		return ledgerState.ResolveOutputState(reference)
-	}, workers, conflictDAG, func(error) {})
+	}, workers, conflictDAG, api.SingleVersionProvider(tpkg.TestAPI), func(error) {})
 
 	tf := mempooltests.NewTestFramework(t, memPoolInstance, conflictDAG, ledgerState, workers)
 
@@ -70,15 +72,51 @@ func TestMempoolV1_ResourceCleanup(t *testing.T) {
 	txIndex, prevStateAlias := issueTransactions(1, 10, "genesis")
 	tf.WaitChildren()
 
-	require.Equal(t, 0, memPoolInstance.cachedTransactions.Size())
+	require.Equal(t, 10, memPoolInstance.cachedTransactions.Size())
+	require.Equal(t, 10, memPoolInstance.delayedTransactionEviction.Size())
 	require.Equal(t, 0, memPoolInstance.stateDiffs.Size())
-	require.Equal(t, 0, memPoolInstance.cachedStateRequests.Size())
+	require.Equal(t, 10, memPoolInstance.delayedOutputStateEviction.Size())
+	// genesis output is never committed or orphaned, so it is not evicted as it has still a non-evicted spender.
+	require.Equal(t, 11, memPoolInstance.cachedStateRequests.Size())
+
+	txIndex, prevStateAlias = issueTransactions(txIndex, 10, prevStateAlias)
+	tf.WaitChildren()
+
+	require.Equal(t, 20, memPoolInstance.cachedTransactions.Size())
+	require.Equal(t, 20, memPoolInstance.delayedTransactionEviction.Size())
+	require.Equal(t, 0, memPoolInstance.stateDiffs.Size())
+	require.Equal(t, 20, memPoolInstance.delayedOutputStateEviction.Size())
+	require.Equal(t, 21, memPoolInstance.cachedStateRequests.Size())
 
 	txIndex, prevStateAlias = issueTransactions(txIndex, 10, prevStateAlias)
 	tf.WaitChildren()
 
+	// Cached transactions stop increasing after 20, because eviction is delayed by MCA.
+	require.Equal(t, 20, memPoolInstance.cachedTransactions.Size())
+	require.Equal(t, 20, memPoolInstance.delayedTransactionEviction.Size())
+	require.Equal(t, 0, memPoolInstance.stateDiffs.Size())
+	require.Equal(t, 20, memPoolInstance.delayedOutputStateEviction.Size())
+	require.Equal(t, 21, memPoolInstance.cachedStateRequests.Size())
+
+	txIndex, _ = issueTransactions(txIndex, 10, prevStateAlias)
+	tf.WaitChildren()
+
+	require.Equal(t, 20, memPoolInstance.cachedTransactions.Size())
+	require.Equal(t, 20, memPoolInstance.delayedTransactionEviction.Size())
+	require.Equal(t, 0, memPoolInstance.stateDiffs.Size())
+	require.Equal(t, 20, memPoolInstance.delayedOutputStateEviction.Size())
+	require.Equal(t, 21, memPoolInstance.cachedStateRequests.Size())
+
+	for index := txIndex; index <= txIndex+20; index++ {
+		tf.CommitSlot(iotago.SlotIndex(index))
+		tf.Instance.Evict(iotago.SlotIndex(index))
+	}
+
+	// Genesis output is also finally evicted thanks to the fact that all its spenders got evicted.
 	require.Equal(t, 0, memPoolInstance.cachedTransactions.Size())
+	require.Equal(t, 0, memPoolInstance.delayedTransactionEviction.Size())
 	require.Equal(t, 0, memPoolInstance.stateDiffs.Size())
+	require.Equal(t, 0, memPoolInstance.delayedOutputStateEviction.Size())
 	require.Equal(t, 0, memPoolInstance.cachedStateRequests.Size())
 
 	attachmentsSlotCount := 0
@@ -105,7 +143,7 @@ func newTestFramework(t *testing.T) *mempooltests.TestFramework {
 
 	return mempooltests.NewTestFramework(t, New[vote.MockedRank](new(mempooltests.VM), func(reference mempool.StateReference) *promise.Promise[mempool.State] {
 		return ledgerState.ResolveOutputState(reference)
-	}, workers, conflictDAG, func(error) {}), conflictDAG, ledgerState, workers)
+	}, workers, conflictDAG, api.SingleVersionProvider(tpkg.TestAPI), func(error) {}), conflictDAG, ledgerState, workers)
 }
 
 func newForkingTestFramework(t *testing.T) *mempooltests.TestFramework {
@@ -116,5 +154,5 @@ func newForkingTestFramework(t *testing.T) *mempooltests.TestFramework {
 
 	return mempooltests.NewTestFramework(t, New[vote.MockedRank](new(mempooltests.VM), func(reference mempool.StateReference) *promise.Promise[mempool.State] {
 		return ledgerState.ResolveOutputState(reference)
-	}, workers, conflictDAG, func(error) {}, WithForkAllTransactions[vote.MockedRank](true)), conflictDAG, ledgerState, workers)
+	}, workers, conflictDAG, api.SingleVersionProvider(tpkg.TestAPI), func(error) {}, WithForkAllTransactions[vote.MockedRank](true)), conflictDAG, ledgerState, workers)
 }
diff --git a/pkg/protocol/engine/mempool/v1/state_metadata.go b/pkg/protocol/engine/mempool/v1/state_metadata.go
index f7f1a6593..80b30f3de 100644
--- a/pkg/protocol/engine/mempool/v1/state_metadata.go
+++ b/pkg/protocol/engine/mempool/v1/state_metadata.go
@@ -4,6 +4,7 @@ import (
 	"sync/atomic"
 
 	"github.com/iotaledger/hive.go/ds/reactive"
+	"github.com/iotaledger/hive.go/lo"
 	"github.com/iotaledger/hive.go/runtime/event"
 	"github.com/iotaledger/hive.go/runtime/promise"
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/mempool"
@@ -53,8 +54,8 @@ func (s *StateMetadata) setup(optSource ...*TransactionMetadata) *StateMetadata
 	source.OnPending(func() { s.accepted.Set(false) })
 	source.OnAccepted(func() { s.accepted.Set(true) })
 	source.OnRejected(func() { s.rejected.Trigger() })
-	source.OnCommitted(func() { s.committed.Trigger() })
-	source.OnOrphaned(func() { s.orphaned.Trigger() })
+	source.OnCommittedSlotUpdated(lo.Void(s.committedSlot.Set))
+	source.OnOrphanedSlotUpdated(lo.Void(s.orphanedSlot.Set))
 
 	return s
 }
@@ -101,8 +102,8 @@ func (s *StateMetadata) AllSpendersRemoved() bool {
 	return s.allSpendersRemoved.WasTriggered()
 }
 
-func (s *StateMetadata) onAllSpendersRemoved(callback func()) (unsubscribe func()) {
-	return s.allSpendersRemoved.Hook(callback).Unhook
+func (s *StateMetadata) onAllSpendersRemoved(callback func()) {
+	s.allSpendersRemoved.Hook(callback)
 }
 
 func (s *StateMetadata) PendingSpenderCount() int {
@@ -138,11 +139,11 @@ func (s *StateMetadata) setupSpender(spender *TransactionMetadata) {
 		s.spendAccepted.Set(nil)
 	})
 
-	spender.OnCommitted(func() {
+	spender.OnCommittedSlotUpdated(func(_ iotago.SlotIndex) {
 		s.spendCommitted.Set(spender)
 
 		s.decreaseSpenderCount()
 	})
 
-	spender.OnOrphaned(s.decreaseSpenderCount)
+	spender.OnOrphanedSlotUpdated(func(_ iotago.SlotIndex) { s.decreaseSpenderCount() })
 }
diff --git a/pkg/protocol/engine/mempool/v1/transaction_metadata.go b/pkg/protocol/engine/mempool/v1/transaction_metadata.go
index 4c49fbb4b..a976dcf92 100644
--- a/pkg/protocol/engine/mempool/v1/transaction_metadata.go
+++ b/pkg/protocol/engine/mempool/v1/transaction_metadata.go
@@ -39,12 +39,13 @@ type TransactionMetadata struct {
 	conflictAccepted      reactive.Event
 
 	// attachments
-	signingTransactions           reactive.Set[*SignedTransactionMetadata]
-	allSigningTransactionsEvicted reactive.Event
+	signingTransactions reactive.Set[*SignedTransactionMetadata]
 
 	validAttachments                *shrinkingmap.ShrinkingMap[iotago.BlockID, bool]
 	earliestIncludedValidAttachment reactive.Variable[iotago.BlockID]
-	allValidAttachmentsEvicted      reactive.Event
+
+	// allValidAttachmentsEvicted is set on the slot of the last and newest evicted attachment
+	allValidAttachmentsEvicted reactive.Variable[iotago.SlotIndex]
 
 	// mutex needed?
 	mutex            syncutils.RWMutex
@@ -84,12 +85,11 @@ func NewTransactionMetadata(transaction mempool.Transaction, referencedInputs []
 		conflicting:           reactive.NewEvent(),
 		conflictAccepted:      reactive.NewEvent(),
 
-		allSigningTransactionsEvicted: reactive.NewEvent(),
-		signingTransactions:           reactive.NewSet[*SignedTransactionMetadata](),
+		signingTransactions: reactive.NewSet[*SignedTransactionMetadata](),
 
 		validAttachments:                shrinkingmap.New[iotago.BlockID, bool](),
 		earliestIncludedValidAttachment: reactive.NewVariable[iotago.BlockID](),
-		allValidAttachmentsEvicted:      reactive.NewEvent(),
+		allValidAttachmentsEvicted:      reactive.NewVariable[iotago.SlotIndex](),
 
 		inclusionFlags: newInclusionFlags(),
 	}).setup(), nil
@@ -215,7 +215,7 @@ func (t *TransactionMetadata) markInputSolid() (allInputsSolid bool) {
 }
 
 func (t *TransactionMetadata) Commit() {
-	t.committed.Trigger()
+	t.committedSlot.Set(t.earliestIncludedValidAttachment.Get().Slot())
 }
 
 func (t *TransactionMetadata) IsConflicting() bool {
@@ -250,7 +250,9 @@ func (t *TransactionMetadata) setupInput(input *StateMetadata) {
 	t.parentConflictIDs.InheritFrom(input.conflictIDs)
 
 	input.OnRejected(func() { t.rejected.Trigger() })
-	input.OnOrphaned(func() { t.orphaned.Trigger() })
+	input.OnOrphanedSlotUpdated(func(slot iotago.SlotIndex) {
+		t.orphanedSlot.Set(slot)
+	})
 	input.OnAccepted(func() {
 		if atomic.AddUint64(&t.unacceptedInputsCount, ^uint64(0)) == 0 {
 			if wereAllInputsAccepted := t.allInputsAccepted.Set(true); !wereAllInputsAccepted {
@@ -268,7 +270,7 @@ func (t *TransactionMetadata) setupInput(input *StateMetadata) {
 	})
 
 	input.OnAcceptedSpenderUpdated(func(spender mempool.TransactionMetadata) {
-		//nolint:forcetypeassert
+		//nolint:forcetypeassert // we can be sure that the spender is a TransactionMetadata
 		if spender.(*TransactionMetadata) != nil && spender != t {
 			t.rejected.Trigger()
 		}
@@ -276,7 +278,9 @@ func (t *TransactionMetadata) setupInput(input *StateMetadata) {
 
 	input.OnSpendCommitted(func(spender mempool.TransactionMetadata) {
 		if spender != t {
-			t.orphaned.Trigger()
+			spender.OnCommittedSlotUpdated(func(slot iotago.SlotIndex) {
+				t.orphanedSlot.Set(slot)
+			})
 		}
 	})
 }
@@ -290,9 +294,9 @@ func (t *TransactionMetadata) setup() (self *TransactionMetadata) {
 		t.conflictIDs.Replace(ds.NewSet(t.id))
 	})
 
-	t.allSigningTransactionsEvicted.OnTrigger(func() {
-		if !t.IsCommitted() {
-			t.orphaned.Trigger()
+	t.allValidAttachmentsEvicted.OnUpdate(func(_, slot iotago.SlotIndex) {
+		if !lo.Return2(t.CommittedSlot()) {
+			t.orphanedSlot.Set(slot)
 		}
 	})
 
@@ -302,9 +306,6 @@ func (t *TransactionMetadata) setup() (self *TransactionMetadata) {
 		}
 	})
 
-	t.OnCommitted(t.setEvicted)
-	t.OnOrphaned(t.setEvicted)
-
 	return t
 }
 
@@ -314,7 +315,7 @@ func (t *TransactionMetadata) addSigningTransaction(signedTransactionMetadata *S
 
 	if added = t.signingTransactions.Add(signedTransactionMetadata); added {
 		signedTransactionMetadata.OnEvicted(func() {
-			t.evictSigningTransaction(signedTransactionMetadata)
+			t.signingTransactions.Delete(signedTransactionMetadata)
 		})
 	}
 
@@ -356,12 +357,6 @@ func (t *TransactionMetadata) evictValidAttachment(id iotago.BlockID) {
 	defer t.attachmentsMutex.Unlock()
 
 	if t.validAttachments.Delete(id) && t.validAttachments.IsEmpty() {
-		t.allValidAttachmentsEvicted.Trigger()
-	}
-}
-
-func (t *TransactionMetadata) evictSigningTransaction(signedTransactionMetadata *SignedTransactionMetadata) {
-	if t.signingTransactions.Delete(signedTransactionMetadata) && t.signingTransactions.IsEmpty() {
-		t.allSigningTransactionsEvicted.Trigger()
+		t.allValidAttachmentsEvicted.Set(id.Slot())
 	}
 }
diff --git a/pkg/tests/booker_test.go b/pkg/tests/booker_test.go
index 15e48560b..6815c8f2f 100644
--- a/pkg/tests/booker_test.go
+++ b/pkg/tests/booker_test.go
@@ -10,6 +10,7 @@ import (
 	"github.com/iotaledger/iota-core/pkg/protocol"
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/blocks"
 	"github.com/iotaledger/iota-core/pkg/testsuite"
+	"github.com/iotaledger/iota-core/pkg/testsuite/mock"
 	iotago "github.com/iotaledger/iota.go/v4"
 )
 
@@ -218,3 +219,422 @@ func Test_MultipleAttachments(t *testing.T) {
 		ts.AssertConflictsInCacheAcceptanceState([]string{"tx1", "tx2"}, acceptance.Accepted, nodeA, nodeB)
 	}
 }
+
+func Test_SpendRejectedCommittedRace(t *testing.T) {
+	ts := testsuite.NewTestSuite(t,
+		testsuite.WithGenesisTimestampOffset(20*10),
+		testsuite.WithMinCommittableAge(2),
+		testsuite.WithMaxCommittableAge(5),
+	)
+	defer ts.Shutdown()
+
+	node1 := ts.AddValidatorNode("node1")
+	node2 := ts.AddValidatorNode("node2")
+
+	ts.Run(true, map[string][]options.Option[protocol.Protocol]{})
+
+	ts.AssertSybilProtectionCommittee(0, []iotago.AccountID{
+		node1.AccountID,
+		node2.AccountID,
+	}, ts.Nodes()...)
+
+	genesisCommitment := lo.PanicOnErr(node1.Protocol.MainEngineInstance().Storage.Commitments().Load(0)).Commitment()
+
+	// Create and issue double spends
+	{
+		tx1 := lo.PanicOnErr(ts.TransactionFramework.CreateSimpleTransaction("tx1", 1, "Genesis:0"))
+		tx2 := lo.PanicOnErr(ts.TransactionFramework.CreateSimpleTransaction("tx2", 1, "Genesis:0"))
+
+		ts.IssueBlockAtSlotWithOptions("block1.1", 1, genesisCommitment, node1, tx1)
+		ts.IssueBlockAtSlotWithOptions("block1.2", 1, genesisCommitment, node1, tx2)
+		ts.IssueBlockAtSlot("block2.tx1", 2, genesisCommitment, node1, ts.BlockIDs("block1.1")...)
+
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+		ts.AssertTransactionsInCacheBooked(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+		ts.AssertTransactionsInCachePending(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block1.1"):   {"tx1"},
+			ts.Block("block1.2"):   {"tx2"},
+			ts.Block("block2.tx1"): {"tx1"},
+		}, node1, node2)
+
+		ts.AssertTransactionInCacheConflicts(map[*iotago.Transaction][]string{
+			ts.TransactionFramework.Transaction("tx2"): {"tx2"},
+			ts.TransactionFramework.Transaction("tx1"): {"tx1"},
+		}, node1, node2)
+	}
+
+	// Issue some more blocks and assert that conflicts are propagated to blocks.
+	{
+		ts.IssueBlockAtSlot("block2.1", 2, genesisCommitment, node1, ts.BlockID("block1.1"))
+		ts.IssueBlockAtSlot("block2.2", 2, genesisCommitment, node1, ts.BlockID("block1.2"))
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block2.1"):   {"tx1"},
+			ts.Block("block2.2"):   {"tx2"},
+			ts.Block("block2.tx1"): {"tx1"},
+		}, node1, node2)
+		ts.AssertTransactionsInCachePending(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+	}
+
+	// Issue valid blocks that resolve the conflict.
+	{
+		ts.IssueBlockAtSlot("block2.3", 2, genesisCommitment, node2, ts.BlockIDs("block2.2")...)
+		ts.IssueBlockAtSlot("block2.4", 2, genesisCommitment, node1, ts.BlockIDs("block2.3")...)
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block2.3"):   {"tx2"},
+			ts.Block("block2.tx1"): {"tx1"},
+		}, node1, node2)
+		ts.AssertTransactionsInCacheAccepted(ts.TransactionFramework.Transactions("tx2"), true, node1, node2)
+		ts.AssertTransactionsInCacheRejected(ts.TransactionFramework.Transactions("tx1"), true, node1, node2)
+	}
+
+	// Advance both nodes at the edge of slot 1 committability
+	{
+		ts.IssueBlocksAtSlots("", []iotago.SlotIndex{2, 3, 4}, 1, "block2.4", ts.Nodes("node1", "node2"), false, nil)
+
+		ts.AssertNodeState(ts.Nodes(),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(0),
+			testsuite.WithEqualStoredCommitmentAtIndex(0),
+			testsuite.WithEvictedSlot(0),
+		)
+
+		ts.IssueBlockAtSlot("block5.1", 5, genesisCommitment, node1, ts.BlockIDsWithPrefix("block1.1")...)
+		ts.IssueBlockAtSlot("block5.2", 5, genesisCommitment, node1, ts.BlockIDsWithPrefix("block1.2")...)
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block5.1"):   {"tx1"}, // on rejected conflict
+			ts.Block("block5.2"):   {},      // accepted merged-to-master
+			ts.Block("block2.tx1"): {"tx1"},
+		}, node1, node2)
+
+		ts.IssueBlocksAtSlots("", []iotago.SlotIndex{5}, 1, "4.0", ts.Nodes("node1"), false, nil)
+
+		ts.AssertBlocksExist(ts.BlocksWithPrefix("5.0"), true, ts.Nodes()...)
+	}
+
+	partitions := map[string][]*mock.Node{
+		"node1": {node1},
+		"node2": {node2},
+	}
+
+	// Split the nodes into partitions and commit slot 1 only on node2
+	{
+
+		ts.SplitIntoPartitions(partitions)
+
+		// Only node2 will commit after issuing this one
+		ts.IssueBlocksAtSlots("", []iotago.SlotIndex{5}, 1, "5.0", ts.Nodes("node2"), false, nil)
+
+		ts.AssertNodeState(ts.Nodes("node1"),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(0),
+			testsuite.WithEqualStoredCommitmentAtIndex(0),
+			testsuite.WithEvictedSlot(0),
+		)
+
+		ts.AssertNodeState(ts.Nodes("node2"),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(1),
+			testsuite.WithEqualStoredCommitmentAtIndex(1),
+			testsuite.WithEvictedSlot(1),
+		)
+	}
+
+	commitment1 := lo.PanicOnErr(node2.Protocol.MainEngineInstance().Storage.Commitments().Load(1)).Commitment()
+
+	// This should be booked on the rejected tx1 conflict
+	tx4 := lo.PanicOnErr(ts.TransactionFramework.CreateSimpleTransaction("tx4", 1, "tx1:0"))
+
+	// Issue TX3 on top of rejected TX1 and 1 commitment on node2 (committed to slot 1)
+	{
+		ts.IssueBlockAtSlotWithOptions("n2-commit1", 5, commitment1, node2, tx4)
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("n2-commit1"): {}, // no conflits inherited as the block is invalid and doesn't get booked.
+			ts.Block("block2.tx1"): {"tx1"},
+		}, node2)
+
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1"), true, node2)
+		ts.AssertTransactionsInCacheRejected(ts.TransactionFramework.Transactions("tx4"), true, node2)
+		ts.AssertTransactionsInCacheBooked(ts.TransactionFramework.Transactions("tx4"), true, node2)
+
+		// As the block commits to 1 but spending something orphaned in 1 it should be invalid
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n2-commit1"), false, node2)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n2-commit1"), true, node2)
+	}
+
+	// Issue a block on node1 that inherits a pending conflict that has been orphaned on node2
+	{
+		ts.IssueBlockAtSlot("n1-rejected-genesis", 5, genesisCommitment, node1, ts.BlockIDs("block2.tx1")...)
+
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n1-rejected-genesis"), true, node1)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n1-rejected-genesis"), false, node1)
+
+		ts.AssertTransactionsInCacheRejected(ts.TransactionFramework.Transactions("tx1"), true, node2)
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block2.tx1"):          {"tx1"},
+			ts.Block("n1-rejected-genesis"): {"tx1"}, // on rejected conflict
+		}, node1)
+	}
+
+	// Issue TX4 on top of rejected TX1 but Genesis commitment on node2 (committed to slot 1)
+	{
+		ts.IssueBlockAtSlotWithOptions("n2-genesis", 5, genesisCommitment, node2, tx4, blockfactory.WithStrongParents(ts.BlockID("Genesis")))
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("n2-genesis"): {"tx4"}, // on rejected conflict
+		}, node2)
+
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n2-genesis"), true, node2)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n2-genesis"), false, node2)
+	}
+
+	// Issue TX4 on top of rejected TX1 but Genesis commitment on node1 (committed to slot 0)
+	{
+		ts.IssueBlockAtSlotWithOptions("n1-genesis", 5, genesisCommitment, node1, tx4, blockfactory.WithStrongParents(ts.BlockID("Genesis")))
+
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1"), true, node2)
+		ts.AssertTransactionsInCacheRejected(ts.TransactionFramework.Transactions("tx4"), true, node2)
+		ts.AssertTransactionsInCacheBooked(ts.TransactionFramework.Transactions("tx4"), true, node2)
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("n1-genesis"): {"tx4"}, // on rejected conflict
+		}, node1)
+
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n1-genesis"), true, node1)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n1-genesis"), false, node1)
+	}
+
+	ts.MergePartitionsToMain(lo.Keys(partitions)...)
+
+	// Sync up the nodes to he same point and check consistency between them.
+	{
+		// Let node1 catch up with commitment 1
+		ts.IssueBlocksAtSlots("5.1", []iotago.SlotIndex{5}, 1, "5.0", ts.Nodes("node2"), false, nil)
+
+		ts.AssertNodeState(ts.Nodes("node1", "node2"),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(1),
+			testsuite.WithEqualStoredCommitmentAtIndex(1),
+			testsuite.WithEvictedSlot(1),
+		)
+
+		// Exchange each-other blocks, ignoring invalidity
+		ts.IssueExistingBlock("n2-genesis", node1)
+		ts.IssueExistingBlock("n2-commit1", node1)
+		ts.IssueExistingBlock("n1-genesis", node2)
+		ts.IssueExistingBlock("n1-rejected-genesis", node2)
+
+		ts.IssueBlockAtSlot("n1-rejected-commit1", 5, commitment1, node1, ts.BlockIDs("n1-rejected-genesis")...)
+		// Needs reissuing on node2 because it is invalid
+		ts.IssueExistingBlock("n1-rejected-commit1", node2)
+
+		// The nodes agree on the results of the invalid blocks
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n2-genesis", "n1-genesis", "n1-rejected-genesis"), true, node1, node2)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n2-genesis", "n1-genesis", "n1-rejected-genesis"), false, node1, node2)
+
+		// This block propagates the orphaned conflict from Tangle
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n1-rejected-commit1"), true, node1, node2)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n1-rejected-commit1"), false, node1, node2)
+
+		// This block spends an orphaned conflict from its Transaction
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n2-commit1"), false, node1, node2)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n2-commit1"), true, node1, node2)
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("n1-genesis"):          {"tx4"}, // on rejected conflict
+			ts.Block("n2-genesis"):          {"tx4"}, // on rejected conflict
+			ts.Block("n1-rejected-genesis"): {"tx1"}, // on rejected conflict
+			ts.Block("n2-commit1"):          {},      // invalid block
+			ts.Block("n1-rejected-commit1"): {},      // merged-to-master
+		}, node1, node2)
+	}
+
+	// Commit further and test eviction of transactions
+	{
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1", "tx2", "tx4"), true, node1, node2)
+
+		ts.IssueBlocksAtSlots("", []iotago.SlotIndex{6, 7, 8, 9, 10}, 5, "5.1", ts.Nodes("node1", "node2"), false, nil)
+
+		ts.AssertNodeState(ts.Nodes("node1", "node2"),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(8),
+			testsuite.WithEqualStoredCommitmentAtIndex(8),
+			testsuite.WithEvictedSlot(8),
+		)
+
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1", "tx2", "tx4"), false, node1, node2)
+	}
+}
+
+func Test_SpendPendingCommittedRace(t *testing.T) {
+	ts := testsuite.NewTestSuite(t,
+		testsuite.WithGenesisTimestampOffset(20*10),
+		testsuite.WithMinCommittableAge(2),
+		testsuite.WithMaxCommittableAge(5),
+	)
+	defer ts.Shutdown()
+
+	node1 := ts.AddValidatorNode("node1")
+	node2 := ts.AddValidatorNode("node2")
+
+	ts.Run(true, map[string][]options.Option[protocol.Protocol]{})
+
+	ts.AssertSybilProtectionCommittee(0, []iotago.AccountID{
+		node1.AccountID,
+		node2.AccountID,
+	}, ts.Nodes()...)
+
+	genesisCommitment := lo.PanicOnErr(node1.Protocol.MainEngineInstance().Storage.Commitments().Load(0)).Commitment()
+
+	// Create and issue double spends
+	{
+		tx1 := lo.PanicOnErr(ts.TransactionFramework.CreateSimpleTransaction("tx1", 1, "Genesis:0"))
+		tx2 := lo.PanicOnErr(ts.TransactionFramework.CreateSimpleTransaction("tx2", 1, "Genesis:0"))
+
+		ts.IssueBlockAtSlotWithOptions("block1.1", 1, genesisCommitment, node2, tx1)
+		ts.IssueBlockAtSlotWithOptions("block1.2", 1, genesisCommitment, node2, tx2)
+
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+		ts.AssertTransactionsInCacheBooked(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+		ts.AssertTransactionsInCachePending(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block1.1"): {"tx1"},
+			ts.Block("block1.2"): {"tx2"},
+		}, node1, node2)
+
+		ts.AssertTransactionInCacheConflicts(map[*iotago.Transaction][]string{
+			ts.TransactionFramework.Transaction("tx2"): {"tx2"},
+			ts.TransactionFramework.Transaction("tx1"): {"tx1"},
+		}, node1, node2)
+	}
+
+	// Issue some more blocks and assert that conflicts are propagated to blocks.
+	{
+		ts.IssueBlockAtSlot("block2.1", 2, genesisCommitment, node2, ts.BlockID("block1.1"))
+		ts.IssueBlockAtSlot("block2.2", 2, genesisCommitment, node2, ts.BlockID("block1.2"))
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block2.1"): {"tx1"},
+			ts.Block("block2.2"): {"tx2"},
+		}, node1, node2)
+		ts.AssertTransactionsInCachePending(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+	}
+
+	// Advance both nodes at the edge of slot 1 committability
+	{
+		ts.IssueBlocksAtSlots("", []iotago.SlotIndex{2, 3, 4}, 1, "Genesis", ts.Nodes("node1", "node2"), false, nil)
+
+		ts.AssertNodeState(ts.Nodes(),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(0),
+			testsuite.WithEqualStoredCommitmentAtIndex(0),
+			testsuite.WithEvictedSlot(0),
+		)
+
+		ts.IssueBlockAtSlot("", 5, genesisCommitment, node1, ts.BlockIDsWithPrefix("4.0")...)
+
+		ts.IssueBlocksAtSlots("", []iotago.SlotIndex{5}, 1, "4.0", ts.Nodes("node1"), false, nil)
+
+		ts.AssertBlocksExist(ts.BlocksWithPrefix("5.0"), true, ts.Nodes()...)
+	}
+
+	partitions := map[string][]*mock.Node{
+		"node1": {node1},
+		"node2": {node2},
+	}
+
+	// Split the nodes into partitions and commit slot 1 only on node2
+	{
+		ts.SplitIntoPartitions(partitions)
+
+		// Only node2 will commit after issuing this one
+		ts.IssueBlocksAtSlots("", []iotago.SlotIndex{5}, 1, "5.0", ts.Nodes("node2"), false, nil)
+
+		ts.AssertNodeState(ts.Nodes("node1"),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(0),
+			testsuite.WithEqualStoredCommitmentAtIndex(0),
+			testsuite.WithEvictedSlot(0),
+		)
+
+		ts.AssertNodeState(ts.Nodes("node2"),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(1),
+			testsuite.WithEqualStoredCommitmentAtIndex(1),
+			testsuite.WithEvictedSlot(1),
+		)
+	}
+
+	commitment1 := lo.PanicOnErr(node2.Protocol.MainEngineInstance().Storage.Commitments().Load(1)).Commitment()
+
+	// Issue a block booked on a pending conflict on node2
+	{
+		ts.IssueBlockAtSlot("n2-pending-genesis", 5, genesisCommitment, node2, ts.BlockIDs("block2.1")...)
+		ts.IssueBlockAtSlot("n2-pending-commit1", 5, commitment1, node2, ts.BlockIDs("block2.1")...)
+
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1"), true, node2)
+		ts.AssertTransactionsInCachePending(ts.TransactionFramework.Transactions("tx1"), true, node2)
+
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n2-pending-genesis", "n2-pending-commit1"), true, node2)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n2-pending-genesis", "n2-pending-commit1"), false, node2)
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block2.1"):           {"tx1"},
+			ts.Block("n2-pending-genesis"): {"tx1"},
+			ts.Block("n2-pending-commit1"): {}, // no conflits inherited as the block merges orphaned conflicts.
+		}, node2)
+	}
+
+	ts.MergePartitionsToMain(lo.Keys(partitions)...)
+
+	// Sync up the nodes to he same point and check consistency between them.
+	{
+		// Let node1 catch up with commitment 1
+		ts.IssueBlocksAtSlots("5.1", []iotago.SlotIndex{5}, 1, "5.0", ts.Nodes("node2"), false, nil)
+
+		ts.AssertNodeState(ts.Nodes("node1", "node2"),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(1),
+			testsuite.WithEqualStoredCommitmentAtIndex(1),
+			testsuite.WithEvictedSlot(1),
+		)
+
+		// Exchange each-other blocks, ignoring invalidity
+		ts.IssueExistingBlock("n2-pending-genesis", node1)
+		ts.IssueExistingBlock("n2-pending-commit1", node1)
+
+		// The nodes agree on the results of the invalid blocks
+		ts.AssertBlocksInCacheBooked(ts.Blocks("n2-pending-genesis", "n2-pending-commit1"), true, node1, node2)
+		ts.AssertBlocksInCacheInvalid(ts.Blocks("n2-pending-genesis", "n2-pending-commit1"), false, node1, node2)
+
+		ts.AssertBlocksInCacheConflicts(map[*blocks.Block][]string{
+			ts.Block("block2.1"):           {"tx1"},
+			ts.Block("n2-pending-genesis"): {"tx1"},
+			ts.Block("n2-pending-commit1"): {}, // no conflits inherited as the block merges orphaned conflicts.
+		}, node1, node2)
+
+		ts.AssertTransactionsInCachePending(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+	}
+
+	// Commit further and test eviction of transactions
+	{
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+		ts.AssertTransactionsInCachePending(ts.TransactionFramework.Transactions("tx1", "tx2"), true, node1, node2)
+
+		ts.IssueBlocksAtSlots("", []iotago.SlotIndex{6, 7, 8, 9, 10}, 5, "5.1", ts.Nodes("node1", "node2"), false, nil)
+
+		ts.AssertNodeState(ts.Nodes("node1", "node2"),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithLatestCommitmentSlotIndex(8),
+			testsuite.WithEqualStoredCommitmentAtIndex(8),
+			testsuite.WithEvictedSlot(8),
+		)
+
+		ts.AssertTransactionsExist(ts.TransactionFramework.Transactions("tx1", "tx2"), false, node1, node2)
+	}
+}
diff --git a/pkg/tests/protocol_startup_test.go b/pkg/tests/protocol_startup_test.go
index 03af7cc9b..2418cfc22 100644
--- a/pkg/tests/protocol_startup_test.go
+++ b/pkg/tests/protocol_startup_test.go
@@ -20,6 +20,98 @@ import (
 	iotago "github.com/iotaledger/iota.go/v4"
 )
 
+func Test_BookInCommittedSlot(t *testing.T) {
+	ts := testsuite.NewTestSuite(t,
+		testsuite.WithLivenessThresholdLowerBound(10),
+		testsuite.WithLivenessThresholdUpperBound(10),
+		testsuite.WithMinCommittableAge(2),
+		testsuite.WithMaxCommittableAge(4),
+		testsuite.WithEpochNearingThreshold(2),
+		testsuite.WithSlotsPerEpochExponent(3),
+		testsuite.WithGenesisTimestampOffset(1000*10),
+	)
+	defer ts.Shutdown()
+
+	nodeA := ts.AddValidatorNode("nodeA")
+
+	nodeOptions := []options.Option[protocol.Protocol]{
+		protocol.WithStorageOptions(
+			storage.WithPruningDelay(20),
+		),
+	}
+
+	ts.Run(true, map[string][]options.Option[protocol.Protocol]{
+		"nodeA": nodeOptions,
+	})
+
+	ts.Wait()
+
+	expectedCommittee := []iotago.AccountID{
+		nodeA.AccountID,
+	}
+
+	expectedOnlineCommittee := []account.SeatIndex{
+		lo.Return1(nodeA.Protocol.MainEngineInstance().SybilProtection.SeatManager().Committee(1).GetSeat(nodeA.AccountID)),
+	}
+
+	// Verify that nodes have the expected states.
+	genesisCommitment := iotago.NewEmptyCommitment(ts.API.ProtocolParameters().Version())
+	genesisCommitment.ReferenceManaCost = ts.API.ProtocolParameters().CongestionControlParameters().MinReferenceManaCost
+	ts.AssertNodeState(ts.Nodes(),
+		testsuite.WithSnapshotImported(true),
+		testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+		testsuite.WithLatestCommitment(genesisCommitment),
+		testsuite.WithLatestFinalizedSlot(0),
+		testsuite.WithChainID(genesisCommitment.MustID()),
+		testsuite.WithStorageCommitments([]*iotago.Commitment{genesisCommitment}),
+		testsuite.WithSybilProtectionCommittee(0, expectedCommittee),
+		testsuite.WithSybilProtectionOnlineCommittee(expectedOnlineCommittee...),
+		testsuite.WithEvictedSlot(0),
+		testsuite.WithActiveRootBlocks(ts.Blocks("Genesis")),
+		testsuite.WithStorageRootBlocks(ts.Blocks("Genesis")),
+	)
+
+	var expectedStorageRootBlocksFrom0 []*blocks.Block
+
+	// Epoch 0: issue 4 rows per slot.
+	{
+		ts.IssueBlocksAtEpoch("", 0, 4, "Genesis", ts.Nodes(), true, nil)
+
+		ts.AssertBlocksExist(ts.BlocksWithPrefixes("1", "2", "3", "4", "5", "6", "7"), true, ts.Nodes()...)
+
+		ts.AssertBlocksInCachePreAccepted(ts.BlocksWithPrefixes("7.3"), true, ts.Nodes()...)
+
+		var expectedActiveRootBlocks []*blocks.Block
+		for _, slot := range []iotago.SlotIndex{3, 4, 5} {
+			expectedActiveRootBlocks = append(expectedActiveRootBlocks, ts.BlocksWithPrefix(fmt.Sprintf("%d.3-", slot))...)
+		}
+
+		for _, slot := range []iotago.SlotIndex{1, 2, 3, 4, 5, 6} {
+			expectedStorageRootBlocksFrom0 = append(expectedStorageRootBlocksFrom0, ts.BlocksWithPrefix(fmt.Sprintf("%d.3-", slot))...)
+		}
+
+		ts.AssertNodeState(ts.Nodes(),
+			testsuite.WithSnapshotImported(true),
+			testsuite.WithProtocolParameters(ts.API.ProtocolParameters()),
+			testsuite.WithChainID(genesisCommitment.MustID()),
+			testsuite.WithLatestCommitmentSlotIndex(5),
+			testsuite.WithEvictedSlot(5),
+			testsuite.WithActiveRootBlocks(expectedActiveRootBlocks),
+			testsuite.WithStorageRootBlocks(expectedStorageRootBlocksFrom0),
+		)
+
+		for _, slot := range []iotago.SlotIndex{4, 5} {
+			aliases := lo.Map([]string{"nodeA"}, func(s string) string {
+				return fmt.Sprintf("%d.3-%s", slot, s)
+			})
+			ts.AssertAttestationsForSlot(slot, ts.Blocks(aliases...), ts.Nodes()...)
+		}
+		ts.IssueBlockAtSlot("5*", 5, lo.PanicOnErr(nodeA.Protocol.MainEngineInstance().Storage.Commitments().Load(3)).Commitment(), ts.Node("nodeA"), ts.BlockIDsWithPrefix("4.3-")...)
+
+		ts.AssertBlocksExist(ts.Blocks("5*"), false, ts.Nodes("nodeA")...)
+	}
+}
+
 func Test_StartNodeFromSnapshotAndDisk(t *testing.T) {
 	ts := testsuite.NewTestSuite(t,
 		testsuite.WithLivenessThresholdLowerBound(10),
diff --git a/pkg/testsuite/blocks.go b/pkg/testsuite/blocks.go
index 225df6b1e..5f80a46a3 100644
--- a/pkg/testsuite/blocks.go
+++ b/pkg/testsuite/blocks.go
@@ -109,6 +109,14 @@ func (t *TestSuite) AssertBlocksInCacheRootBlock(expectedBlocks []*blocks.Block,
 	t.assertBlocksInCacheWithFunc(expectedBlocks, "root-block", expectedRootBlock, (*blocks.Block).IsRootBlock, nodes...)
 }
 
+func (t *TestSuite) AssertBlocksInCacheBooked(expectedBlocks []*blocks.Block, expectedBooked bool, nodes ...*mock.Node) {
+	t.assertBlocksInCacheWithFunc(expectedBlocks, "booked", expectedBooked, (*blocks.Block).IsBooked, nodes...)
+}
+
+func (t *TestSuite) AssertBlocksInCacheInvalid(expectedBlocks []*blocks.Block, expectedInvalid bool, nodes ...*mock.Node) {
+	t.assertBlocksInCacheWithFunc(expectedBlocks, "valid", expectedInvalid, (*blocks.Block).IsInvalid, nodes...)
+}
+
 func (t *TestSuite) AssertBlocksInCacheConflicts(blockConflicts map[*blocks.Block][]string, nodes ...*mock.Node) {
 	for _, node := range nodes {
 		for block, conflictAliases := range blockConflicts {
@@ -135,7 +143,6 @@ func (t *TestSuite) AssertBlocksInCacheConflicts(blockConflicts map[*blocks.Bloc
 
 				return nil
 			})
-
 		}
 	}
 }
diff --git a/pkg/testsuite/mock/node.go b/pkg/testsuite/mock/node.go
index ec7615611..3d62cb05c 100644
--- a/pkg/testsuite/mock/node.go
+++ b/pkg/testsuite/mock/node.go
@@ -413,12 +413,12 @@ func (n *Node) attachEngineLogs(failOnBlockFiltered bool, instance *engine.Engin
 			fmt.Printf("%s > [%s] MemPool.TransactionInvalid(%s): %s\n", n.Name, engineName, err, transactionMetadata.ID())
 		})
 
-		transactionMetadata.OnOrphaned(func() {
-			fmt.Printf("%s > [%s] MemPool.TransactionOrphaned: %s\n", n.Name, engineName, transactionMetadata.ID())
+		transactionMetadata.OnOrphanedSlotUpdated(func(slot iotago.SlotIndex) {
+			fmt.Printf("%s > [%s] MemPool.TransactiOnOrphanedSlotUpdated in slot %d: %s\n", n.Name, engineName, slot, transactionMetadata.ID())
 		})
 
-		transactionMetadata.OnCommitted(func() {
-			fmt.Printf("%s > [%s] MemPool.TransactionCommitted: %s\n", n.Name, engineName, transactionMetadata.ID())
+		transactionMetadata.OnCommittedSlotUpdated(func(slot iotago.SlotIndex) {
+			fmt.Printf("%s > [%s] MemPool.TransactiOnCommittedSlotUpdated in slot %d: %s\n", n.Name, engineName, slot, transactionMetadata.ID())
 		})
 
 		transactionMetadata.OnPending(func() {
@@ -508,6 +508,12 @@ func (n *Node) IssueBlock(ctx context.Context, alias string, opts ...options.Opt
 	return block
 }
 
+func (n *Node) IssueExistingBlock(block *blocks.Block) {
+	require.NoErrorf(n.Testing, n.blockIssuer.IssueBlock(block.ModelBlock()), "%s > failed to issue block with alias %s", n.Name, block.ID().Alias())
+
+	fmt.Printf("%s > Issued block: %s - slot %d - commitment %s %d - latest finalized slot %d\n", n.Name, block.ID(), block.ID().Slot(), block.SlotCommitmentID(), block.SlotCommitmentID().Slot(), block.ProtocolBlock().LatestFinalizedSlot)
+}
+
 func (n *Node) IssueValidationBlock(ctx context.Context, alias string, opts ...options.Option[blockfactory.ValidatorBlockParams]) *blocks.Block {
 	block := n.CreateValidationBlock(ctx, alias, opts...)
 
diff --git a/pkg/testsuite/testsuite_issue_blocks.go b/pkg/testsuite/testsuite_issue_blocks.go
index 5ee05af51..9a1941226 100644
--- a/pkg/testsuite/testsuite_issue_blocks.go
+++ b/pkg/testsuite/testsuite_issue_blocks.go
@@ -94,6 +94,17 @@ func (t *TestSuite) IssueBlockAtSlot(alias string, slot iotago.SlotIndex, slotCo
 	return block
 }
 
+func (t *TestSuite) IssueExistingBlock(alias string, node *mock.Node) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	block, exists := t.blocks.Get(alias)
+	require.True(t.Testing, exists)
+	require.NotNil(t.Testing, block)
+
+	node.IssueExistingBlock(block)
+}
+
 func (t *TestSuite) IssueValidationBlockWithOptions(alias string, node *mock.Node, blockOpts ...options.Option[blockfactory.ValidatorBlockParams]) *blocks.Block {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
@@ -270,7 +281,6 @@ func (t *TestSuite) SlotsForEpoch(epoch iotago.EpochIndex) []iotago.SlotIndex {
 }
 
 func (t *TestSuite) CommitUntilSlot(slot iotago.SlotIndex, activeNodes []*mock.Node, parent *blocks.Block) *blocks.Block {
-
 	// we need to get accepted tangle time up to slot + minCA + 1
 	// first issue a chain of blocks with step size minCA up until slot + minCA + 1
 	// then issue one more block to accept the last in the chain which will trigger commitment of the second last in the chain
diff --git a/pkg/testsuite/transactions.go b/pkg/testsuite/transactions.go
index 12977d5c5..65cedd452 100644
--- a/pkg/testsuite/transactions.go
+++ b/pkg/testsuite/transactions.go
@@ -125,6 +125,10 @@ func (t *TestSuite) AssertTransactionsInCachePending(expectedTransactions []*iot
 	t.assertTransactionsInCacheWithFunc(expectedTransactions, expectedFlag, mempool.TransactionMetadata.IsPending, nodes...)
 }
 
+func (t *TestSuite) AssertTransactionsInCacheOrphaned(expectedTransactions []*iotago.Transaction, expectedFlag bool, nodes ...*mock.Node) {
+	t.assertTransactionsInCacheWithFunc(expectedTransactions, expectedFlag, func(tm mempool.TransactionMetadata) bool { return lo.Return2(tm.OrphanedSlot()) }, nodes...)
+}
+
 func (t *TestSuite) AssertTransactionInCacheConflicts(transactionConflicts map[*iotago.Transaction][]string, nodes ...*mock.Node) {
 	for _, node := range nodes {
 		for transaction, conflictAliases := range transactionConflicts {