Some cleanups and implementation of client into the write strategy

Author: 0x19

.go-version

@@ -0,0 +1 @@
+go1.24.3

TODO.md

@@ -0,0 +1,60 @@
+# FDB TODO List
+
+## Core Production Features
+
+### Record Versioning and Conflict Resolution
+
+- Add version tracking to the `Record` structure:
+  ```go
+  // Record represents a key-value pair in a record batch
+  type Record struct {
+      Key       [32]byte // Fixed-size byte array for keys
+      Value     []byte   // Value as byte slice
+      Version   uint64   // Monotonically increasing version number
+      Priority  uint8    // Priority level for processing
+      Timestamp int64    // Creation/modification timestamp
+  }
+  ```
+
+- Implement conflict detection and resolution mechanisms:
+  - Last-write-wins strategy based on version numbers
+  - Optimistic concurrency control for client operations
+  - Support for conditional writes based on expected versions
+
+- Update serialization/deserialization methods to handle new fields
+
+### Missing Production-Ready Features
+
+1. **Consensus Protocol**
+   - Implement strong consensus for write ordering
+   - Consider Raft or PBFT for distributed agreement
+
+2. **Transaction Management**
+   - Implement ACID transaction support
+   - Add distributed commit/rollback protocols
+   - Support for multi-key atomic operations
+
+3. **Failure Detection and Recovery**
+   - Automated node failure detection
+   - Efficient state transfer for recovering nodes
+   - Automatic recovery procedures
+
+4. **State Transfer Optimization**
+   - Efficient incremental state transfer
+   - Snapshot-based recovery for new/rejoining nodes
+
+5. **Comprehensive Security**
+   - Data encryption (at-rest and in-transit)
+   - Enhanced authentication mechanisms
+   - Audit logging for all operations
+
+6. **Backup & Disaster Recovery**
+   - Point-in-time recovery
+   - Incremental backup support
+   - Disaster recovery procedures
+
+## Performance Optimizations
+
+- Further optimize BatchWriter for higher throughput
+- Improve worker selection algorithm for P2P distribution
+- Enhance TCP streaming for large payloads

client/client.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"github.com/unpackdev/fdb/pkg/types"
 	"sync"
 	"time"
 )
@@ -44,6 +45,17 @@ func (c *Client) GetTransport(name string) (Transport, error) {
 	return transport, nil
 }
 
+// GetTransportByType retrieves a registered transport by type
+func (c *Client) GetTransportByType(name types.TransportType) (Transport, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	transport, exists := c.transports[name.String()]
+	if !exists {
+		return nil, fmt.Errorf("transport by type not found: %s", name)
+	}
+	return transport, nil
+}
+
 // SendMessage sends a message using the specified transport
 func (c *Client) SendMessage(name string, data []byte) error {
 	transport, err := c.GetTransport(name)
@@ -89,49 +101,49 @@ func (c *Client) SendAndReceiveMessage(name string, data []byte, timeout time.Du
 	if err != nil {
 		return nil, err
 	}
-	
+
 	// Check if we're dealing with a TCPTransport which has advanced response handling
 	tcpTransport, ok := transport.(*TCPTransport)
 	if !ok {
 		return nil, fmt.Errorf("SendAndReceiveMessage is only supported for TCP transport, got %T", transport)
 	}
-	
+
 	// Get the response message type - use the success status
 	responseType := MessageType(1) // Using 1 as a default success value
-	
+
 	// Register a response channel before sending the message
 	responseCh := tcpTransport.RegisterResponseChannel(responseType)
-	
+
 	// Define the maximum size for non-chunked messages
 	const maxChunkSize = 1024 * 1024 // 1MB
-	
+
 	// For large payloads, use chunking protocol to ensure reliable transmission
 	encodedMsg := data
 	if len(data) > maxChunkSize {
 		// Prepare the chunked message with a 4-byte length prefix
 		// The server expects: [total_size(4 bytes)][payload...]
 		lengthPrefix := make([]byte, 4)
 		binary.LittleEndian.PutUint32(lengthPrefix, uint32(len(data)))
-		
+
 		// Prepend the length prefix to the message
 		chunkedMsg := append(lengthPrefix, data...)
-		
+
 		// Replace the original message with the chunked version
 		encodedMsg = chunkedMsg
 	}
-	
+
 	// Send the message
 	if err := tcpTransport.Send(encodedMsg); err != nil {
 		// Make sure to unregister the response channel on error
 		tcpTransport.UnregisterResponseChannel(responseType)
 		return nil, fmt.Errorf("failed to send message: %w", err)
 	}
-	
+
 	// Wait for the response with the provided timeout
 	response, err := tcpTransport.WaitForResponseWithTimeout(responseCh, timeout)
 	if err != nil {
 		return nil, fmt.Errorf("error waiting for response: %w", err)
 	}
-	
+
 	return response, nil
 }

config.yaml

@@ -128,7 +128,7 @@ rpc:
 pprof:
   - name: fdb
     enabled: true
-    addr: "0.0.0.0:6060"
+    addr: "0.0.0.0:6065"
 
 # ======================================================================================
 # OBSERVABILITY CONFIGURATION (Metrics and Tracing)

playground/strategies/write.go

@@ -7,6 +7,8 @@ import (
 	"time"
 
 	"github.com/unpackdev/fdb/pkg/logger"
+	"github.com/unpackdev/fdb/pkg/messages"
+	"github.com/unpackdev/fdb/pkg/types"
 	"github.com/unpackdev/fdb/playground/suite"
 )
 
@@ -109,12 +111,14 @@ func WithTotalOperations(ops int) WriteStrategyOption {
 
 // Start begins executing the write strategy
 func (s *WriteStrategy) Start(ctx context.Context) error {
-	s.logger.Info("Starting write strategy",
+	s.logger.Info(
+		"Starting write strategy",
 		"workers", s.workersCount,
 		"data_size_kb", s.dataSizeKB,
 		"ops_per_sec", s.opsPerSec,
 		"total_ops", s.totalOps,
-		"target_node", s.targetNode.PeerID().String())
+		"target_node", s.targetNode.PeerID().String(),
+	)
 
 	ctx, s.cancel = context.WithCancel(ctx)
 	s.startTime = time.Now()
@@ -209,20 +213,53 @@ func (s *WriteStrategy) runWorker(ctx context.Context, id, ops int, delay time.D
 
 // performWriteOperation executes a single write operation
 func (s *WriteStrategy) performWriteOperation(ctx context.Context, workerID, opID int) error {
-	key := fmt.Sprintf("%s%d-%d", s.keyPrefix, workerID, opID)
+	//key := fmt.Sprintf("%s%d-%d", s.keyPrefix, workerID, opID)
 
 	data, err := suite.GenerateTestDataKB(s.dataSizeKB, false)
 	if err != nil {
 		return fmt.Errorf("failed to generate test data: %w", err)
 	}
 
-	// TODO: Implement actual write operation using the client
-	// This is a placeholder for the actual implementation that would use the
-	// optimized BatchWriter component with its 2048 batch size and 100ms flush interval
+	key, err := messages.GenerateRandomKey()
+	if err != nil {
+		return fmt.Errorf("failed to generate random key: %w", err)
+	}
 
-	s.logger.Debug("Write operation",
+	// Create a write message with the key and data
+	writeMsg := &messages.Message{
+		Handler: types.WriteHandlerType,
+		Key:     key,
+		Data:    data,
+	}
+
+	// Encode the message
+	encodedWriteMsg, err := writeMsg.Encode()
+	if err != nil {
+		return fmt.Errorf("failed to encode write message: %w", err)
+	}
+
+	// Send the message to the target node using the client
+	// This leverages the optimized BatchWriter component with 2048 batch size and 100ms flush interval
+	// that's already configured in the underlying implementation
+	response, err := s.targetNode.SendAndReceiveMessage(
+		ctx,
+		s.targetNode,
+		types.TCPTransportType,
+		types.WriteHandlerType,
+		encodedWriteMsg,
+		5*time.Second,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to write record to target node: %w", err)
+	}
+
+	s.logger.Debug(
+		"Write operation successful",
 		"key", key,
-		"data_size", len(data))
+		"data_size", len(data),
+		"worker_id", workerID,
+		"response", response,
+	)
 
 	return nil
 }