Feat: double spends get elevated to their own reality

packages/ledgerstate/ledgerstate_test.go

@@ -19,7 +19,6 @@ var (
 	addressHash3   = NewAddressHash("ADDRESS3")
 	addressHash4   = NewAddressHash("ADDRESS4")
 	pendingReality = NewRealityId("PENDING")
-	conflictingReality = NewRealityId("CONFLICTING")
 )
 
 func Benchmark(b *testing.B) {
@@ -70,7 +69,9 @@ func Test(t *testing.T) {
 		addressHash4, NewColoredBalance(eth, 1000),
 	))
 
-	time.Sleep(3 * time.Second)
+	time.Sleep(100 * time.Millisecond)
+
+	objectstorage.WaitForWritesToFlush()
 
 	ledgerState.ForEachTransferOutput(func(object *objectstorage.CachedObject) bool {
 		object.Consume(func(object objectstorage.StorableObject) {

packages/ledgerstate/reality.go

@@ -130,7 +130,41 @@ func (reality *Reality) BookTransfer(transfer *Transfer) error {
 	return reality.bookTransfer(transfer.GetHash(), reality.ledgerState.getTransferInputs(transfer), transfer.GetOutputs())
 }
 
-func (reality *Reality) consumeInputs(inputs []*objectstorage.CachedObject, transferHash TransferHash, outputs map[AddressHash][]*ColoredBalance) (conflicting bool, err error) {
+func (reality *Reality) elevateTransferOutput(transferOutputReference *TransferOutputReference, newRealityId RealityId) error {
+	cachedTransferOutputToElevate := reality.ledgerState.GetTransferOutput(transferOutputReference)
+	if !cachedTransferOutputToElevate.Exists() {
+		return errors.New("could not find TransferOutput to elevate")
+	}
+
+	transferOutputToElevate := cachedTransferOutputToElevate.Get().(*TransferOutput)
+
+	if transferOutputToElevate.GetRealityId() == reality.id {
+		if err := transferOutputToElevate.moveToReality(newRealityId); err != nil {
+			panic(err)
+		}
+		cachedTransferOutputToElevate.Store()
+		cachedTransferOutputToElevate.Release()
+	} else {
+		fmt.Println("ALREADY ELEVATED")
+	}
+
+	return nil
+}
+
+// Creates a new reality for consumers that have previously been booked in this reality.
+func (reality *Reality) elevateTransferOutputConsumersToOwnReality(consumers map[TransferHash][]AddressHash) {
+	for transferHash, addressHashes := range consumers {
+		var elevatedRealityId RealityId
+		copy(elevatedRealityId[:], transferHash[:])
+		reality.CreateReality(elevatedRealityId).Release()
+
+		for _, addressHash := range addressHashes {
+			reality.elevateTransferOutput(NewTransferOutputReference(transferHash, addressHash), elevatedRealityId)
+		}
+	}
+}
+
+func (reality *Reality) consumeInputs(inputs objectstorage.CachedObjects, transferHash TransferHash, outputs map[AddressHash][]*ColoredBalance) (conflicting bool, err error) {
 	for _, input := range inputs {
 		consumedTransferOutput := input.Get().(*TransferOutput)
 
@@ -141,20 +175,19 @@ func (reality *Reality) consumeInputs(inputs []*objectstorage.CachedObject, tran
 			return
 		}
 
-		if inputConflicting {
-			for transferHash, consumerToElevate := range consumersToElevate {
-				// elevate previous liked transactions
-				fmt.Println(transferHash, consumerToElevate)
-			}
+		if inputConflicting && len(consumersToElevate) >= 1 {
+			reality.elevateTransferOutputConsumersToOwnReality(consumersToElevate)
 		}
 
 		conflicting = conflicting || inputConflicting
+
+		input.Store()
 	}
 
 	return
 }
 
-func (reality *Reality) bookTransfer(transferHash TransferHash, inputs []*objectstorage.CachedObject, outputs map[AddressHash][]*ColoredBalance) error {
+func (reality *Reality) bookTransfer(transferHash TransferHash, inputs objectstorage.CachedObjects, outputs map[AddressHash][]*ColoredBalance) error {
 	if err := reality.checkTransferBalances(inputs, outputs); err != nil {
 		return err
 	}
@@ -176,6 +209,7 @@ func (reality *Reality) bookTransfer(transferHash TransferHash, inputs []*object
 	}
 
 	// 5. release objects
+	inputs.Release()
 
 	return nil
 }

packages/ledgerstate/transfer_output.go

@@ -91,7 +91,7 @@ func (transferOutput *TransferOutput) addConsumer(consumer TransferHash, outputs
 		}
 		consumers := make([]AddressHash, len(outputs))
 		i := 0
-		for addressHash, _ := range outputs {
+		for addressHash := range outputs {
 			consumers[i] = addressHash
 
 			i++
@@ -113,6 +113,10 @@ func (transferOutput *TransferOutput) moveToReality(realityId RealityId) error {
 
 		return err
 	} else {
+		transferOutput.realityIdMutex.Lock()
+		transferOutput.realityId = realityId
+		transferOutput.realityIdMutex.Unlock()
+
 		transferOutput.ledgerState.storeTransferOutputBooking(newTransferOutputBooking(realityId, transferOutput.addressHash, len(transferOutput.consumers) >= 1, transferOutput.transferHash)).Release()
 
 		oldTransferOutputBooking.Delete().Release()
@@ -164,7 +168,15 @@ func (transferOutput *TransferOutput) MarshalBinary() ([]byte, error) {
 	balanceCount := len(transferOutput.balances)
 	consumerCount := len(transferOutput.consumers)
 
-	result := make([]byte, realityIdLength+4+balanceCount*coloredBalanceLength+4+consumerCount*transferHashLength)
+	serializedLength := realityIdLength + 4 + balanceCount*coloredBalanceLength + 4 + consumerCount*transferHashLength
+	for _, addresses := range transferOutput.consumers {
+		serializedLength += 4
+		for range addresses {
+			serializedLength += addressHashLength
+		}
+	}
+
+	result := make([]byte, serializedLength)
 
 	copy(result[0:], transferOutput.realityId[:])
 
@@ -173,8 +185,8 @@ func (transferOutput *TransferOutput) MarshalBinary() ([]byte, error) {
 		copy(result[realityIdLength+4+i*coloredBalanceLength:], transferOutput.balances[i].color[:colorLength])
 		binary.LittleEndian.PutUint64(result[realityIdLength+4+i*coloredBalanceLength+colorLength:], transferOutput.balances[i].balance)
 	}
-
 	offset := realityIdLength + 4 + balanceCount*coloredBalanceLength
+
 	binary.LittleEndian.PutUint32(result[offset:], uint32(consumerCount))
 	offset += 4
 	for transferHash, addresses := range transferOutput.consumers {

packages/ledgerstate/transfer_output_booking.go

@@ -3,6 +3,7 @@ package ledgerstate
 import (
 	"github.com/iotaledger/goshimmer/packages/errors"
 	"github.com/iotaledger/goshimmer/packages/objectstorage"
+	"github.com/iotaledger/goshimmer/packages/stringify"
 )
 
 // region struct + constructor + public api ////////////////////////////////////////////////////////////////////////////
@@ -45,6 +46,15 @@ func (booking *TransferOutputBooking) GetTransferHash() TransferHash {
 	return booking.transferHash
 }
 
+func (booking *TransferOutputBooking) String() string {
+	return stringify.Struct("TransferOutputBooking",
+		stringify.StructField("realityId", booking.realityId),
+		stringify.StructField("addressHash", booking.addressHash),
+		stringify.StructField("spent", booking.spent),
+		stringify.StructField("transferHash", booking.transferHash),
+	)
+}
+
 // endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 // region support object storage ///////////////////////////////////////////////////////////////////////////////////////

packages/ledgerstate/transfer_output_test.go

+package ledgerstate
+
+import (
+	"testing"
+
+	"github.com/magiconair/properties/assert"
+)
+
+func TestTransferOutput_MarshalUnmarshal(t *testing.T) {
+	transferOutput := NewTransferOutput(nil, NewRealityId("REALITY"), NewTransferHash("RECEIVE"), NewAddressHash("ADDRESS1"), NewColoredBalance(NewColor("IOTA"), 44), NewColoredBalance(NewColor("BTC"), 88))
+	transferOutput.consumers = make(map[TransferHash][]AddressHash)
+
+	spendTransferHash := NewTransferHash("SPEND")
+	transferOutput.consumers[spendTransferHash] = make([]AddressHash, 2)
+	transferOutput.consumers[spendTransferHash][0] = NewAddressHash("ADDRESS2")
+	transferOutput.consumers[spendTransferHash][1] = NewAddressHash("ADDRESS3")
+
+	marshaledData, err := transferOutput.MarshalBinary()
+	if err != nil {
+		t.Error(err)
+	}
+
+	unmarshaledTransferOutput := TransferOutput{storageKey: transferOutput.GetStorageKey()}
+	if err := unmarshaledTransferOutput.UnmarshalBinary(marshaledData); err != nil {
+		t.Error(err)
+	}
+
+	assert.Equal(t, unmarshaledTransferOutput.realityId, transferOutput.realityId)
+	assert.Equal(t, unmarshaledTransferOutput.transferHash, transferOutput.transferHash)
+	assert.Equal(t, unmarshaledTransferOutput.addressHash, transferOutput.addressHash)
+	assert.Equal(t, len(unmarshaledTransferOutput.consumers), len(transferOutput.consumers))
+	assert.Equal(t, len(unmarshaledTransferOutput.consumers[spendTransferHash]), len(transferOutput.consumers[spendTransferHash]))
+	assert.Equal(t, unmarshaledTransferOutput.consumers[spendTransferHash][0], transferOutput.consumers[spendTransferHash][0])
+	assert.Equal(t, unmarshaledTransferOutput.consumers[spendTransferHash][1], transferOutput.consumers[spendTransferHash][1])
+}

packages/ledgerstate/types.go

-package ledgerstate
-
-// empty structs consume 0 memory and are therefore better in map based set implementations that i.e. bool
-type empty struct{}
-
-// void represents an empty value that consumes 0 memory
-var void empty

packages/objectstorage/batch_writer.go

@@ -12,6 +12,10 @@ import (
 
 var writeWg sync.WaitGroup
 
+var timeoutWg sync.WaitGroup
+
+var waitingForTimeout bool
+
 var startStopMutex sync.Mutex
 
 var running int32 = 0
@@ -38,6 +42,10 @@ func StopBatchWriter() {
 	startStopMutex.Unlock()
 }
 
+func WaitForWritesToFlush() {
+	timeoutWg.Wait()
+}
+
 func batchWrite(object *CachedObject) {
 	if atomic.LoadInt32(&running) == 0 {
 		StartBatchWriter()
@@ -84,6 +92,11 @@ func runBatchWriter() {
 	for atomic.LoadInt32(&running) == 1 {
 		writeWg.Add(1)
 
+		if !waitingForTimeout {
+			waitingForTimeout = true
+			timeoutWg.Add(1)
+		}
+
 		wb := badgerInstance.NewWriteBatch()
 
 		writtenValues := make([]*CachedObject, BATCH_WRITER_BATCH_SIZE)
@@ -97,6 +110,9 @@ func runBatchWriter() {
 				writtenValues[writtenValuesCounter] = objectToPersist
 				writtenValuesCounter++
 			case <-time.After(BATCH_WRITER_BATCH_TIMEOUT):
+				waitingForTimeout = false
+				timeoutWg.Done()
+
 				break COLLECT_VALUES
 			case <-daemon.ShutdownSignal:
 				break COLLECT_VALUES

packages/objectstorage/cached_object.go

@@ -57,9 +57,7 @@ func (cachedObject *CachedObject) Release() {
 
 // Directly consumes the StorableObject. This method automatically Release()s the object when the callback is done.
 func (cachedObject *CachedObject) Consume(consumer func(object StorableObject)) {
-	if cachedObject.IsDeleted() {
-		consumer(nil)
-	} else if cachedObject.Exists() {
+	if cachedObject.Exists() && !cachedObject.IsDeleted() {
 		consumer(cachedObject.Get())
 	}
 

packages/objectstorage/cached_objects.go

+package objectstorage
+
+type CachedObjects []*CachedObject
+
+func (cachedObjects CachedObjects) Release() {
+	for _, cachedObject := range cachedObjects {
+		cachedObject.Release()
+	}
+}
+
+func (cachedObjects CachedObjects) Store() {
+	for _, cachedObject := range cachedObjects {
+		cachedObject.Store()
+	}
+}