@@ -43,7 +43,7 @@ func TestP2PDistribution(t *testing.T) {
4343 require .NoError (t , err , "Failed to shutdown test nodes" )
4444 }()
4545
46- // Send a test message and wait for response
46+ // Verify the cluster is up and running
4747 testMessage := []byte ("test message payload" )
4848 resp , err := nodes .GetBootstrapNode ().SendAndReceiveMessage (
4949 nodes .GetBootstrapNode (),
@@ -54,120 +54,120 @@ func TestP2PDistribution(t *testing.T) {
5454 )
5555 require .NoError (t , err , "Failed to send message or receive response" )
5656 require .NotNil (t , resp , "Response should not be nil" )
57-
58- // The write handler responds with a 1-byte response: 0x00 for success or 0x01 for error
59- // See transports/db_write_handler.go for details
6057 require .Equal (t , 1 , len (resp ), "Response should be exactly 1 byte" )
6158 require .Equal (t , byte (0x00 ), resp [0 ], "Response should be 0x00 (success)" )
6259
63- // Test P2P distributor
64- t .Run ("P2P Distribution" , func (t * testing.T ) {
65- // Create a test record with a fixed key for consistency
66- // Generate a random 32-byte key
67- key , err := messages .GenerateRandomKey ()
68- require .NoError (t , err , "Failed to generate random key" )
69-
70- value := []byte ("test record value payload" )
71-
72- t .Logf ("Using test key (hex): %s" , hex .EncodeToString (key [:]))
73-
74- // Get the bootstrap node and a regular node
75- bootstrapNode := nodes .GetBootstrapNode ()
76- regularNode := nodes .GetNodeByIndex (1 ) // Get the second node
77- require .NotNil (t , regularNode , "Failed to get the second node" )
78-
79- // Create a message with the fixed key
80- t .Log ("Writing record to bootstrap node database" )
81-
82- // Create a message with our specific key (not random)
83- writeMsg := & messages.Message {
84- Handler : types .WriteHandlerType ,
85- Key : key ,
86- Data : value ,
87- }
88-
89- // Encode the message
90- encodedWriteMsg , err := writeMsg .Encode ()
91- require .NoError (t , err , "Failed to encode write message" )
92-
93- // Write the record to the bootstrap node
94- writeResp , err := bootstrapNode .SendAndReceiveMessage (
95- bootstrapNode ,
96- types .TCPTransportType ,
97- types .WriteHandlerType ,
98- encodedWriteMsg ,
99- 5 * time .Second ,
100- )
101- require .NoError (t , err , "Failed to write record to bootstrap node" )
102- require .Equal (t , byte (0x00 ), writeResp [0 ], "Write operation should succeed with 0x00" )
103-
104- // The DbWriteHandler automatically distributes records after writing
105- // We're explicitly calling a BatchWriter.Flush() in the handler
106-
107- // Give some time for distribution to occur
108- t .Log ("Waiting for record distribution..." )
109- time .Sleep (1 * time .Second )
110-
111- // Verify on bootstrap node first (should be immediate)
112- t .Log ("Verifying record exists on bootstrap node" )
113-
114- // Create a read message with the SAME key
115- t .Logf ("Reading with the exact same key (hex): %s" , hex .EncodeToString (key [:]))
116-
117- // Create a read message with our specific key (not random)
118- readMsg := & messages.Message {
119- Handler : types .ReadHandlerType ,
120- Key : key , // Same key we used for writing
121- Data : nil , // No data needed for read requests
122- }
123-
124- // Encode the message
125- encodedReadMsg , err := readMsg .Encode ()
126- require .NoError (t , err , "Failed to encode read message" )
127-
128- // Read from the bootstrap node
129- bootstrapResp , err := bootstrapNode .SendAndReceiveMessage (
130- bootstrapNode ,
131- types .TCPTransportType ,
132- types .ReadHandlerType ,
133- encodedReadMsg ,
134- 5 * time .Second ,
135- )
136- require .NoError (t , err , "Failed to read record from bootstrap node" )
137- require .NotNil (t , bootstrapResp , "Bootstrap node response should not be nil" )
138- require .True (t , len (bootstrapResp ) > 1 , "Bootstrap node response should have at least a status byte and value" )
139-
140- // Response format is now: [1 byte status][value bytes]
141- // For successful reads, status byte should be 0x00
142- require .Equal (t , byte (0x00 ), bootstrapResp [0 ], "Read response status byte should be 0x00" )
143-
144- // Extract the actual payload (skip the status byte)
145- bootstrapValue := bootstrapResp [1 :]
146- require .Equal (t , value , bootstrapValue , "Record value mismatch on bootstrap node" )
147-
148- // Now verify on the regular node (should get there via P2P distribution)
149- t .Log ("Verifying record exists on regular node" )
150-
151- // Read from the regular node using the same message
152- regularResp , err := regularNode .SendAndReceiveMessage (
153- regularNode ,
154- types .TCPTransportType ,
155- types .ReadHandlerType ,
156- encodedReadMsg ,
157- 5 * time .Second ,
158- )
159- require .NoError (t , err , "Failed to read record from regular node" )
160- require .NotNil (t , regularResp , "Regular node response should not be nil" )
161- require .True (t , len (regularResp ) > 1 , "Regular node response should have at least a status byte and value" )
162-
163- // Response format is now: [1 byte status][value bytes]
164- // For successful reads, status byte should be 0x00
165- require .Equal (t , byte (0x00 ), regularResp [0 ], "Read response status byte should be 0x00" )
166-
167- // Extract the actual payload (skip the status byte)
168- regularValue := regularResp [1 :]
169- require .Equal (t , value , regularValue , "Record value mismatch on regular node" )
170-
171- t .Log ("P2P distribution test successful!" )
172- })
60+ // Get all nodes we'll use for testing
61+ bootstrapNode := nodes .GetBootstrapNode ()
62+ regularNode := nodes .GetNodeByIndex (1 )
63+ require .NotNil (t , regularNode , "Failed to get second node" )
64+
65+ // Define table-driven test cases
66+ testCases := []struct {
67+ name string
68+ value []byte
69+ waitTime time.Duration // Time to wait for P2P distribution
70+ }{
71+ {
72+ name : "Basic string value" ,
73+ value : []byte ("test record value payload" ),
74+ waitTime : 100 * time .Millisecond ,
75+ },
76+ {
77+ name : "JSON data" ,
78+ value : []byte (`{"id":"12345","name":"test","data":[1,2,3,4,5]}` ),
79+ waitTime : 100 * time .Millisecond ,
80+ },
81+ // This is broken, 4096 bytes is currently processing fine, need to get it working by buffering...
82+ // {
83+ // name: "Large binary data (100KB)",
84+ // value: bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 25 * 1024), // ~100KB of data
85+ // waitTime: 3 * time.Second, // Extra time for larger payload
86+ // },
87+ }
88+
89+ // Run all test cases
90+ for _ , tc := range testCases {
91+ t .Run (tc .name , func (t * testing.T ) {
92+ // Generate a unique key for this test case
93+ key , err := messages .GenerateRandomKey ()
94+ require .NoError (t , err , "Failed to generate random key" )
95+ t .Logf ("Using test key (hex): %s" , hex .EncodeToString (key [:]))
96+
97+ // 1. WRITE PHASE: Write record to bootstrap node
98+ t .Log ("Writing record to bootstrap node database" )
99+ writeMsg := & messages.Message {
100+ Handler : types .WriteHandlerType ,
101+ Key : key ,
102+ Data : tc .value ,
103+ }
104+
105+ encodedWriteMsg , err := writeMsg .Encode ()
106+ require .NoError (t , err , "Failed to encode write message" )
107+
108+ writeResp , err := bootstrapNode .SendAndReceiveMessage (
109+ bootstrapNode ,
110+ types .TCPTransportType ,
111+ types .WriteHandlerType ,
112+ encodedWriteMsg ,
113+ 5 * time .Second ,
114+ )
115+ require .NoError (t , err , "Failed to write record to bootstrap node" )
116+ require .Equal (t , byte (0x00 ), writeResp [0 ], "Write operation should succeed with 0x00" )
117+
118+ // Wait for P2P distribution to occur
119+ // Batch writter flushes immediately if there's > bufferSize messages or every 100ms it flushes
120+ // so let's await for a little bit of the time (100ms is enough)
121+ t .Logf ("Waiting %v for record distribution..." , tc .waitTime )
122+ time .Sleep (tc .waitTime )
123+
124+ // 2. READ PHASE: Create read message for verification
125+ readMsg := & messages.Message {
126+ Handler : types .ReadHandlerType ,
127+ Key : key ,
128+ Data : nil ,
129+ }
130+ encodedReadMsg , err := readMsg .Encode ()
131+ require .NoError (t , err , "Failed to encode read message" )
132+
133+ // 2a. Verify on bootstrap node (should be immediate)
134+ t .Log ("Verifying record exists on bootstrap node" )
135+ bootstrapResp , err := bootstrapNode .SendAndReceiveMessage (
136+ bootstrapNode ,
137+ types .TCPTransportType ,
138+ types .ReadHandlerType ,
139+ encodedReadMsg ,
140+ 5 * time .Second ,
141+ )
142+ require .NoError (t , err , "Failed to read record from bootstrap node" )
143+ require .NotNil (t , bootstrapResp , "Bootstrap node response should not be nil" )
144+ require .True (t , len (bootstrapResp ) > 1 , "Bootstrap node response should have at least a status byte and value" )
145+ require .Equal (t , byte (0x00 ), bootstrapResp [0 ], "Read response status byte should be 0x00" )
146+
147+ // Extract the actual payload (skip the status byte)
148+ bootstrapValue := bootstrapResp [1 :]
149+ require .Equal (t , tc .value , bootstrapValue , "Record value mismatch on bootstrap node" )
150+
151+ // 2b. Verify on regular node (should get there via P2P distribution)
152+ t .Log ("Verifying record exists on regular node" )
153+ regularResp , err := regularNode .SendAndReceiveMessage (
154+ regularNode ,
155+ types .TCPTransportType ,
156+ types .ReadHandlerType ,
157+ encodedReadMsg ,
158+ 5 * time .Second ,
159+ )
160+ require .NoError (t , err , "Failed to read record from regular node" )
161+ require .NotNil (t , regularResp , "Regular node response should not be nil" )
162+ require .True (t , len (regularResp ) > 1 , "Regular node response should have at least a status byte and value" )
163+ require .Equal (t , byte (0x00 ), regularResp [0 ], "Read response status byte should be 0x00" )
164+
165+ // Extract the actual payload (skip the status byte)
166+ regularValue := regularResp [1 :]
167+ require .Equal (t , tc .value , regularValue , "Record value mismatch on regular node" )
168+
169+ t .Logf ("%s: P2P distribution successful!" , tc .name )
170+ })
171+ }
172+
173173}
0 commit comments