[coordinator-test] Remove the need for a real VC

Signed-off-by: Liran Funaro <liran.funaro@gmail.com>

Author: liran-funaro

Makefile

@@ -132,7 +132,7 @@ MAKEFLAGS += --jobs=16
 
 ROOT_PKG_REGEXP = github.com/hyperledger/fabric-x-committer
 CORE_DB_PACKAGES_REGEXP = ${ROOT_PKG_REGEXP}/service/(vc|query)
-REQUIRES_DB_PACKAGES_REGEXP = ${ROOT_PKG_REGEXP}/(service/coordinator|loadgen|cmd|utils/testdb)
+REQUIRES_DB_PACKAGES_REGEXP = ${ROOT_PKG_REGEXP}/(loadgen|cmd|utils/testdb)
 HEAVY_PACKAGES_REGEXP = ${ROOT_PKG_REGEXP}/(docker|integration)
 
 NON_HEAVY_PACKAGES=$(shell $(go_cmd) list ./... | grep -vE "$(HEAVY_PACKAGES_REGEXP)")

mock/fifo_cache.go

@@ -20,7 +20,7 @@ func newFifoCache[T any](size int) *fifoCache[T] {
 	}
 }
 
-// addIfNotExist returns true if the entry is unique.
+// addIfNotExist returns true if a new entry was added.
 // It employs a FIFO eviction policy.
 func (c *fifoCache[T]) addIfNotExist(key string, value T) bool {
 	if _, ok := c.get(key); ok {
@@ -33,6 +33,17 @@ func (c *fifoCache[T]) addIfNotExist(key string, value T) bool {
 	return true
 }
 
+// updateOrAddIfNotExist returns true if a new entry was added.
+// It employs a FIFO eviction policy.
+func (c *fifoCache[T]) updateOrAddIfNotExist(key string, value T) bool {
+	if _, ok := c.get(key); ok {
+		c.cache[key] = value
+		// We don't modify the eviction order for simplicity.
+		return false
+	}
+	return c.addIfNotExist(key, value)
+}
+
 // get returns the value of a key.
 // If the key does not exist, it returns false.
 func (c *fifoCache[T]) get(key string) (T, bool) {

mock/vcservice.go

@@ -17,6 +17,7 @@ import (
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/health"
 	healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	"github.com/hyperledger/fabric-x-committer/api/servicepb"
@@ -32,27 +33,45 @@ type (
 	VcService struct {
 		servicepb.ValidationAndCommitServiceServer
 		streamStateManager[VCStreamState]
-		nextBlock   atomic.Pointer[servicepb.BlockRef]
-		txsStatus   *fifoCache[*committerpb.TxStatus]
-		txsStatusMu sync.Mutex
-		healthcheck *health.Server
+		nextBlock    atomic.Pointer[servicepb.BlockRef]
+		txsStatus    *fifoCache[*committerpb.TxStatus]
+		worldState   *fifoCache[*WorldState]
+		worldStateMu sync.Mutex
+		healthcheck  *health.Server
 		// NumBatchesReceived is the number of batches received by VcService.
 		NumBatchesReceived atomic.Uint32
 		// MockFaultyNodeDropSize allows mocking a faulty node by dropping some TXs.
 		MockFaultyNodeDropSize int
+		// FullMVCC forces the mock VC to perform full MVCC validation.
+		FullMVCC atomic.Bool
 	}
 
 	// VCStreamState holds the stream's batch queue.
 	VCStreamState struct {
 		StreamInfo
 		q chan *servicepb.VcBatch
 	}
+
+	// WorldState describe the world-state of a key.
+	WorldState struct {
+		Namespace string
+		Key       []byte
+		Value     []byte
+		Version   uint64
+	}
+
+	read struct {
+		ns      string
+		key     []byte
+		version *uint64
+	}
 )
 
 // NewMockVcService returns a new VcService.
 func NewMockVcService() *VcService {
 	return &VcService{
 		txsStatus:   newFifoCache[*committerpb.TxStatus](defaultTxStatusStorageSize),
+		worldState:  newFifoCache[*WorldState](defaultTxStatusStorageSize),
 		healthcheck: serve.DefaultHealthCheckService(),
 	}
 }
@@ -103,8 +122,8 @@ func (v *VcService) GetTransactionsStatus(
 	query *committerpb.TxIDsBatch,
 ) (*committerpb.TxStatusBatch, error) {
 	s := &committerpb.TxStatusBatch{Status: make([]*committerpb.TxStatus, len(query.TxIds))}
-	v.txsStatusMu.Lock()
-	defer v.txsStatusMu.Unlock()
+	v.worldStateMu.Lock()
+	defer v.worldStateMu.Unlock()
 	for i, id := range query.TxIds {
 		if status, ok := v.txsStatus.get(id); ok {
 			s.Status[i] = status
@@ -179,26 +198,8 @@ func (v *VcService) sendTransactionStatus(
 		if !ok {
 			break
 		}
-		txsStatus := &committerpb.TxStatusBatch{
-			Status: make([]*committerpb.TxStatus, 0, len(txBatch.Transactions)),
-		}
-		v.txsStatusMu.Lock()
-		for i, tx := range txBatch.Transactions {
-			if i < v.MockFaultyNodeDropSize {
-				// We simulate a faulty node by not responding to the first X TXs.
-				continue
-			}
-			code := committerpb.Status_COMMITTED
-			if tx.PrelimInvalidTxStatus != nil {
-				code = *tx.PrelimInvalidTxStatus
-			}
-			s := committerpb.NewTxStatusFromRef(tx.Ref, code)
-			txsStatus.Status = append(txsStatus.Status, s)
-			v.txsStatus.addIfNotExist(tx.Ref.TxId, s)
-		}
-		v.txsStatusMu.Unlock()
-
-		if err := stream.Send(txsStatus); err != nil {
+		status := v.process(txBatch.Transactions)
+		if err := stream.Send(&committerpb.TxStatusBatch{Status: status}); err != nil {
 			return errors.Wrap(err, "error sending transaction status")
 		}
 	}
@@ -218,3 +219,123 @@ func (v *VcService) SubmitTransactions(ctx context.Context, txsBatch *servicepb.
 	channel.NewWriter(ctx, s.q).Write(txsBatch)
 	return nil
 }
+
+// GetKeys returns the WorldState of the given keys.
+func (v *VcService) GetKeys(nsID string, keys ...[]byte) []*WorldState {
+	res := make([]*WorldState, 0, len(keys))
+	v.worldStateMu.Lock()
+	defer v.worldStateMu.Unlock()
+	for _, k := range keys {
+		if val, ok := v.worldState.get(worldStateKey(nsID, k)); ok {
+			res = append(res, val)
+		}
+	}
+	return res
+}
+
+func (v *VcService) process(txs []*servicepb.VcTx) []*committerpb.TxStatus {
+	status := make([]*committerpb.TxStatus, 0, len(txs))
+
+	// We simulate a faulty node by not responding to the first X TXs.
+	skip := max(0, min(v.MockFaultyNodeDropSize, len(txs)))
+
+	v.worldStateMu.Lock()
+	defer v.worldStateMu.Unlock()
+	for _, tx := range txs[skip:] {
+		code := v.validate(tx)
+		if code == committerpb.Status_COMMITTED {
+			for _, w := range getWrites(tx) {
+				key := worldStateKey(w.Namespace, w.Key)
+				if val, valExist := v.worldState.get(key); valExist {
+					w.Version = val.Version + 1
+				}
+				v.worldState.updateOrAddIfNotExist(key, w)
+			}
+		}
+		s := committerpb.NewTxStatusFromRef(tx.Ref, code)
+		status = append(status, s)
+		v.txsStatus.addIfNotExist(tx.Ref.TxId, s)
+	}
+
+	return status
+}
+
+func (v *VcService) validate(tx *servicepb.VcTx) committerpb.Status {
+	code := committerpb.Status_COMMITTED
+	if tx.PrelimInvalidTxStatus != nil {
+		return *tx.PrelimInvalidTxStatus
+	}
+
+	if !v.FullMVCC.Load() {
+		return code
+	}
+
+	existingStatus, txIDExist := v.txsStatus.get(tx.Ref.TxId)
+	if txIDExist {
+		if proto.Equal(existingStatus.Ref, tx.Ref) {
+			return existingStatus.Status
+		}
+		return committerpb.Status_REJECTED_DUPLICATE_TX_ID
+	}
+
+	for _, r := range getReads(tx) {
+		key := worldStateKey(r.ns, r.key)
+		val, valExist := v.worldState.get(key)
+		if (r.version == nil && valExist) || (r.version != nil && (!valExist || *r.version != val.Version)) {
+			return committerpb.Status_ABORTED_MVCC_CONFLICT
+		}
+	}
+
+	return code
+}
+
+func getReads(tx *servicepb.VcTx) (reads []read) {
+	for _, ns := range tx.Namespaces {
+		if ns.NsId != committerpb.MetaNamespaceID && ns.NsId != committerpb.ConfigNamespaceID {
+			reads = append(reads, read{
+				ns:      committerpb.MetaNamespaceID,
+				key:     []byte(ns.NsId),
+				version: &ns.NsVersion,
+			})
+		}
+		for _, r := range ns.ReadsOnly {
+			reads = append(reads, read{
+				ns:      ns.NsId,
+				key:     r.Key,
+				version: r.Version,
+			})
+		}
+		for _, rw := range ns.ReadWrites {
+			reads = append(reads, read{
+				ns:      ns.NsId,
+				key:     rw.Key,
+				version: rw.Version,
+			})
+		}
+	}
+	return reads
+}
+
+func getWrites(tx *servicepb.VcTx) (writes []*WorldState) {
+	for _, ns := range tx.Namespaces {
+		for _, rw := range ns.ReadWrites {
+			writes = append(writes, &WorldState{
+				Namespace: ns.NsId,
+				Key:       rw.Key,
+				Value:     rw.Value,
+			})
+		}
+		for _, w := range ns.BlindWrites {
+			writes = append(writes, &WorldState{
+				Namespace: ns.NsId,
+				Key:       w.Key,
+				Value:     w.Value,
+			})
+		}
+	}
+	return writes
+}
+
+func worldStateKey(nsID string, key []byte) string {
+	return nsID + "$" + string(key)
+}