Feat: ledger nearing its final design

packages/refactored/ledger/availabilitymanager.go

@@ -1,6 +1,7 @@
 package ledger
 
 import (
+	"github.com/iotaledger/hive.go/generics/dataflow"
 	"github.com/iotaledger/hive.go/generics/event"
 
 	"github.com/iotaledger/goshimmer/packages/refactored/utxo"
@@ -22,29 +23,25 @@ func NewAvailabilityManager(ledger *Ledger) (newAvailabilityManager *Availabilit
 	return newAvailabilityManager
 }
 
-func (a *AvailabilityManager) CheckSolidity(transaction utxo.Transaction, metadata *TransactionMetadata) (inputs []utxo.Output) {
-	if metadata.Solid() {
+func (a *AvailabilityManager) CheckSolidity(params *DataFlowParams, next dataflow.Next[*DataFlowParams]) (err error) {
+	if params.TransactionMetadata.Solid() {
 		return nil
 	}
 
-	solid, inputs := a.allInputsAvailable(transaction)
+	solid, inputs := a.allInputsAvailable(params.Transaction)
 	if !solid {
 		return nil
 	}
 
-	if !metadata.SetSolid(true) {
+	if !params.TransactionMetadata.SetSolid(true) {
 		return nil
 	}
 
-	a.TransactionSolidEvent.Trigger(&TransactionSolidEvent{
-		Inputs: inputs,
-		TransactionStoredEvent: &TransactionStoredEvent{
-			Transaction:         transaction,
-			TransactionMetadata: metadata,
-		},
-	})
+	a.TransactionSolidEvent.Trigger(&TransactionSolidEvent{params})
 
-	return inputs
+	params.Inputs = inputs
+
+	return next(params)
 }
 
 func (a *AvailabilityManager) allInputsAvailable(transaction utxo.Transaction) (allAvailable bool, outputs []utxo.Output) {

packages/refactored/ledger/events.go

@@ -1,20 +1,13 @@
 package ledger
 
-import (
-	"github.com/iotaledger/hive.go/generics/walker"
-
-	"github.com/iotaledger/goshimmer/packages/refactored/utxo"
-)
-
 type TransactionStoredEvent struct {
-	Transaction         utxo.Transaction
-	TransactionMetadata *TransactionMetadata
+	*DataFlowParams
 }
 
 type TransactionSolidEvent struct {
-	Inputs []utxo.Output
-
-	propagationWalker *walker.Walker[utxo.TransactionID]
+	*DataFlowParams
+}
 
-	*TransactionStoredEvent
+type TransactionProcessedEvent struct {
+	*DataFlowParams
 }

packages/refactored/ledger/ledger.go

@@ -1,6 +1,7 @@
 package ledger
 
 import (
+	"github.com/cockroachdb/errors"
 	"github.com/iotaledger/hive.go/generics/dataflow"
 	"github.com/iotaledger/hive.go/generics/event"
 	"github.com/iotaledger/hive.go/generics/walker"
@@ -13,11 +14,11 @@ import (
 // region Ledger ///////////////////////////////////////////////////////////////////////////////////////////////////////
 
 type Ledger struct {
-	ErrorEvent *event.Event[error]
+	ErrorEvent                *event.Event[error]
+	TransactionProcessedEvent *event.Event[*TransactionProcessedEvent]
 
 	*Storage
 	*AvailabilityManager
-	*DataFlow
 	syncutils.DAGMutex[[32]byte]
 
 	vm utxo.VM
@@ -33,52 +34,85 @@ func New(store kvstore.KVStore, vm utxo.VM) (ledger *Ledger) {
 	return ledger
 }
 
-func (l *Ledger) StoreAndProcessTransaction(transaction utxo.Transaction) {
-	cachedTransactionMetadata, stored := l.Store(transaction)
-	defer cachedTransactionMetadata.Release()
-
-	if !stored {
-		return
-	}
+func (l *Ledger) Setup() {
+	// Attach = run async
+	l.TransactionProcessedEvent.Attach(event.NewClosure[*TransactionProcessedEvent](l.ProcessFutureCone))
+}
 
-	l.TransactionStoredEvent.Trigger(&TransactionStoredEvent{
-		Transaction:         transaction,
-		TransactionMetadata: cachedTransactionMetadata.Get(),
+func (l *Ledger) ProcessTransaction(tx utxo.Transaction, meta *TransactionMetadata) (success bool, err error) {
+	err = dataflow.New[*DataFlowParams](
+		l.lockTransactionStep,
+		l.CheckSolidity,
+		/*
+			l.ValidatePastCone,
+			l.ExecuteTransaction,
+			l.BookTransaction,
+		*/
+		l.triggerProcessedEventStep,
+	).WithSuccessCallback(func(params *DataFlowParams) {
+		success = true
+	}).WithTerminationCallback(func(params *DataFlowParams) {
+		l.Unlock(params.Transaction, false)
+	}).Run(&DataFlowParams{
+		Transaction:         tx,
+		TransactionMetadata: meta,
 	})
 
-	l.ProcessTransaction(transaction, cachedTransactionMetadata.Get())
+	if err != nil {
+		// TODO: mark Transaction as invalid and trigger invalid event
+		// eventually trigger generic errors if its not due to tx invalidity
+		return false, err
+	}
+
+	return success, nil
 }
 
-func (l *Ledger) ProcessTransaction(tx utxo.Transaction, meta *TransactionMetadata) {
-	success, consumers, err := l.SolidifyTransaction(tx, meta)
-	if !success {
-		return
-	}
+func (l *Ledger) ProcessFutureCone(event *TransactionProcessedEvent) {
+	// TODO: FILL WITH ACTUAL CONSUMERS
+	_ = event.Inputs
+	consumers := []utxo.TransactionID{}
 
-	// TODO: async event
-	consumersWalker := walker.New[utxo.TransactionID](true).PushAll(consumers...)
-	for consumersWalker.HasNext() {
-		l.CachedTransactionMetadata(consumersWalker.Next()).Consume(func(consumerMetadata *TransactionMetadata) {
-			l.CachedTransaction(consumerMetadata.ID()).Consume(func(consumerTransaction utxo.Transaction) {
-				_, consumers, err := l.SolidifyTransaction(tx, meta)
+	for consumersWalker := walker.New[utxo.TransactionID](true).PushAll(consumers...); consumersWalker.HasNext(); {
+		transactionID := consumersWalker.Next()
 
+		l.CachedTransactionMetadata(transactionID).Consume(func(consumerMetadata *TransactionMetadata) {
+			l.CachedTransaction(transactionID).Consume(func(consumerTransaction utxo.Transaction) {
+				if _, err := l.ProcessTransaction(consumerTransaction, consumerMetadata); err != nil {
+					l.ErrorEvent.Trigger(errors.Errorf("failed to process Transaction with %s: %w", transactionID, err))
+				}
 			})
 		})
+	}
+}
+
+func (l *Ledger) StoreAndProcessTransaction(transaction utxo.Transaction) (success bool, err error) {
+	cachedTransactionMetadata, stored := l.Store(transaction)
+	defer cachedTransactionMetadata.Release()
 
-		consumersWalker.Next()
+	if !stored {
+		return
 	}
 
-	return
+	l.TransactionStoredEvent.Trigger(&TransactionStoredEvent{&DataFlowParams{
+		Transaction:         transaction,
+		TransactionMetadata: nil, // todo retrieve
+	}})
+
+	return l.ProcessTransaction(transaction, cachedTransactionMetadata.Get())
+}
+
+func (l *Ledger) lockTransactionStep(params *DataFlowParams, next dataflow.Next[*DataFlowParams]) error {
+	l.Lock(params.Transaction, false)
+
+	return next(params)
 }
 
-func (l *Ledger) forkSingleTransactionCommand() (solidificationCommand dataflow.ChainedCommand[*DataFlowParams]) {
-	return l.LockedCommand(
-		dataflow.New[*DataFlowParams](
-			l.ForkTransaction,
-		).WithErrorCallback(func(err error, params *DataFlowParams) {
-			// trigger generic errors if its not due to tx invalidity
-		}).ChainedCommand,
-	)
+func (l *Ledger) triggerProcessedEventStep(params *DataFlowParams, next dataflow.Next[*DataFlowParams]) (err error) {
+	l.TransactionProcessedEvent.Trigger(&TransactionProcessedEvent{
+		params,
+	})
+
+	return next(params)
 }
 
 // endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////////