Bunch of changes and bump of mdbx

Author: 0x19

TODO.md

@@ -58,3 +58,5 @@
 - 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,8 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"github.com/google/uuid"
+	"github.com/unpackdev/fdb/pkg/messages"
 	"github.com/unpackdev/fdb/pkg/types"
 	"sync"
 	"time"
@@ -91,13 +93,10 @@ func (c *Client) Close() error {
 	return nil
 }
 
-// SendAndReceiveMessage sends a message to another node and waits for a
-// response. This implementation handles large payloads with chunking and
-// provides proper response handling with timeouts.
-// Note: Currently this method only works with TCPTransport and will return
-// an error for other transport types.
-func (c *Client) SendAndReceiveMessage(name string, data []byte, timeout time.Duration) ([]byte, error) {
-	transport, err := c.GetTransport(name)
+// SendAndReceiveMessage sends a message and waits for a response with the specified timeout.
+func (c *Client) SendAndReceiveMessage(ctx context.Context, transportType string, data []byte, timeout time.Duration) ([]byte, error) {
+	// Get the transport
+	transport, err := c.GetTransport(transportType)
 	if err != nil {
 		return nil, err
 	}
@@ -108,11 +107,23 @@ func (c *Client) SendAndReceiveMessage(name string, data []byte, timeout time.Du
 		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
+	// Try to decode the message to extract the UUID
+	// If it's not already an encoded message, we'll need to generate a new UUID
+	var messageID uuid.UUID
+
+	if decodedMsg, err := messages.Decode(data); err == nil {
+		// Data is already an encoded message, extract the ID
+		messageID = decodedMsg.ID
+	} else {
+		// Generate a new UUID for this request
+		messageID = uuid.New()
+		// Note: we should ideally encode this ID into the data, but for now
+		// we'll just use it for correlation. In a complete implementation,
+		// you'd want to encode the data with this ID.
+	}
 
-	// Register a response channel before sending the message
-	responseCh := tcpTransport.RegisterResponseChannel(responseType)
+	// Register a response channel using the message ID
+	responseCh := tcpTransport.RegisterResponseChannel(messageID)
 
 	// Define the maximum size for non-chunked messages
 	const maxChunkSize = 1024 * 1024 // 1MB
@@ -135,7 +146,7 @@ func (c *Client) SendAndReceiveMessage(name string, data []byte, timeout time.Du
 	// Send the message
 	if err := tcpTransport.Send(encodedMsg); err != nil {
 		// Make sure to unregister the response channel on error
-		tcpTransport.UnregisterResponseChannel(responseType)
+		tcpTransport.UnregisterResponseChannel(messageID)
 		return nil, fmt.Errorf("failed to send message: %w", err)
 	}
 

client/client_tcp_test.go

@@ -4,16 +4,19 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/panjf2000/gnet/v2"
 	"github.com/unpackdev/fdb/pkg/packets"
 	"github.com/unpackdev/fdb/pkg/types"
 	"go.uber.org/zap"
 )
 
+// Global transport instance for default handlers - will be set during registration
+var defaultTransport *TCPTransport
+
 // RegisterDefaultHandlers registers default message type handlers for common error conditions
 // and system messages that the client should always be able to handle.
 func RegisterDefaultHandlers(transport *TCPTransport) {
 	transport.logger.Info("Registering default message handlers for client")
+	defaultTransport = transport
 
 	// Register handler for InvalidActionMessageType (69 decimal = 'E' ASCII)
 	// This message type is sent by the server when it encounters issues with very large payloads
@@ -40,13 +43,12 @@ func RegisterDefaultHandlers(transport *TCPTransport) {
 
 // handleInvalidAction processes InvalidActionMessageType (0x69) messages from the server
 // These typically indicate resource constraints when processing large payloads
-func handleInvalidAction(conn gnet.Conn, data []byte) error {
-	// Try to parse the response data into a DBResponse object
-	dbResponse, err := packets.DecodeDBResponse(data)
+func handleInvalidAction(data []byte) error {
+	// Try to parse the response data into a MessageResponse object
+	dbResponse, err := packets.DecodeMessageResponse(data)
 	if err != nil {
-		handler, ok := conn.Context().(*tcpEventHandler)
-		if ok && handler != nil && handler.transport != nil && handler.transport.logger != nil {
-			handler.transport.logger.Error("Failed to decode InvalidAction response",
+		if defaultTransport != nil && defaultTransport.logger != nil {
+			defaultTransport.logger.Error("Failed to decode InvalidAction response",
 				zap.Error(err),
 				zap.Int("data_length", len(data)))
 		}
@@ -60,46 +62,44 @@ func handleInvalidAction(conn gnet.Conn, data []byte) error {
 	}
 
 	// Log the error
-	handler, ok := conn.Context().(*tcpEventHandler)
-	if ok && handler != nil && handler.transport != nil && handler.transport.logger != nil {
-		handler.transport.logger.Warn("Server reported invalid action or resource constraint",
+	if defaultTransport != nil && defaultTransport.logger != nil {
+		defaultTransport.logger.Warn("Server reported invalid action or resource constraint",
 			zap.String("error", errorMessage),
 			zap.Uint32("data_length", dbResponse.Length),
 			zap.Uint8("status", uint8(dbResponse.Status)))
 	}
 
 	// Since this is a special error type, we can notify any waiting response channels
 	// by converting this to a regular error response
-	errorResp := &packets.DBResponse{
+	errorResp := &packets.MessageResponse{
 		Status: types.HandlerStatusError,
 		Length: uint32(len(errorMessage)),
 		Data:   []byte(errorMessage),
 	}
 
 	// Forward this error to any success message handlers that might be waiting
-	if handler != nil && handler.transport != nil && handler.transport.responseHandler != nil {
+	if defaultTransport != nil && defaultTransport.responseHandler != nil {
 		// Create a new message that response handlers can understand (success message type)
 		successMsgType := MessageType(types.HandlerStatusSuccess.Byte())
 
 		// Encode the error response to be handled
 		encodedResp := errorResp.Encode()
 
 		// Use the response handler's existing mechanism to handle this
-		handler.transport.responseHandler.HandleResponse(successMsgType, encodedResp)
+		defaultTransport.responseHandler.HandleResponse(successMsgType, encodedResp)
 	}
 
 	return nil
 }
 
 // handleErrorResponse processes HandlerStatusError (0) messages from the server
-// These are properly formatted protocol error messages using DBResponse structure
-func handleErrorResponse(conn gnet.Conn, data []byte) error {
-	// Try to parse the response data into a DBResponse object
-	dbResponse, err := packets.DecodeDBResponse(data)
+// These are properly formatted protocol error messages using MessageResponse structure
+func handleErrorResponse(data []byte) error {
+	// Try to parse the response data into a MessageResponse object
+	dbResponse, err := packets.DecodeMessageResponse(data)
 	if err != nil {
-		handler, ok := conn.Context().(*tcpEventHandler)
-		if ok && handler != nil && handler.transport != nil && handler.transport.logger != nil {
-			handler.transport.logger.Error("Failed to decode Error response",
+		if defaultTransport != nil && defaultTransport.logger != nil {
+			defaultTransport.logger.Error("Failed to decode Error response",
 				zap.Error(err),
 				zap.Int("data_length", len(data)))
 		}
@@ -130,20 +130,19 @@ func handleErrorResponse(conn gnet.Conn, data []byte) error {
 	}
 
 	// Log the error
-	handler, ok := conn.Context().(*tcpEventHandler)
-	if ok && handler != nil && handler.transport != nil && handler.transport.logger != nil {
-		handler.transport.logger.Error("Received protocol-formatted error from server",
+	if defaultTransport != nil && defaultTransport.logger != nil {
+		defaultTransport.logger.Error("Received protocol-formatted error from server",
 			zap.String("error_message", errorMessage),
 			zap.Uint32("data_length", dbResponse.Length))
 	}
 
 	// Forward this error to any success message handlers that might be waiting
-	if handler != nil && handler.transport != nil && handler.transport.responseHandler != nil {
+	if defaultTransport != nil && defaultTransport.responseHandler != nil {
 		// Create a new message that response handlers can understand (success message type)
 		successMsgType := MessageType(types.HandlerStatusSuccess.Byte())
 
 		// Just pass the original data - it's already properly formatted
-		handler.transport.responseHandler.HandleResponse(successMsgType, data)
+		defaultTransport.responseHandler.HandleResponse(successMsgType, data)
 	}
 
 	return nil