test(broadcast): added test for ordering

Author: aleksander-vedvik

broadcast/request.go

@@ -211,6 +211,7 @@ func (r *BroadcastRequest) dispatchOutOfOrderMsgs(ctx context.Context) {
 	if r.executionOrder == nil || len(r.executionOrder) <= 0 {
 		return
 	}
+	handledMethods := make([]string, 0, len(r.outOfOrderMsgs))
 	for method, msgs := range r.outOfOrderMsgs {
 		order, ok := r.executionOrder[method]
 		if !ok {
@@ -222,6 +223,11 @@ func (r *BroadcastRequest) dispatchOutOfOrderMsgs(ctx context.Context) {
 			for _, msg := range msgs {
 				msg.Run(ctx)
 			}
+			handledMethods = append(handledMethods, method)
 		}
 	}
+	// cleanup after dispatching the cached messages
+	for _, m := range handledMethods {
+		delete(r.outOfOrderMsgs, m)
+	}
 }

tests/broadcast/broadcast.pb.go

@@ -49,6 +49,18 @@ service BroadcastService {
   rpc GetVal(Request) returns (Response) { 
     option (gorums.broadcastcall) = true; 
   }
+  rpc Order(Request) returns (Response) { 
+    option (gorums.broadcastcall) = true; 
+  }
+  rpc PrePrepare(Request) returns (Empty) { 
+    option (gorums.broadcast) = true; 
+  }
+  rpc Prepare(Request) returns (Empty) { 
+    option (gorums.broadcast) = true; 
+  }
+  rpc Commit(Request) returns (Empty) { 
+    option (gorums.broadcast) = true; 
+  }
 }
 
 message Request { 

tests/broadcast/broadcast.proto

@@ -52,7 +52,7 @@ func createSrvs(numSrvs int, down ...int) ([]*testServer, []string, func(), erro
 func TestSimpleBroadcastCall(t *testing.T) {
 	numSrvs := 3
 	numReqs := 10
-	srvs, srvAddrs, srvCleanup, err := createSrvs(numSrvs)
+	_, srvAddrs, srvCleanup, err := createSrvs(numSrvs)
 	if err != nil {
 		t.Error(err)
 	}
@@ -77,11 +77,6 @@ func TestSimpleBroadcastCall(t *testing.T) {
 		}
 		cancel()
 	}
-	for _, srv := range srvs {
-		if srv.GetNumMsgs() != numReqs*numSrvs {
-			//t.Error(fmt.Sprintf("resp is wrong, want: %v, got: %v", numReqs*numSrvs, srv.GetNumMsgs()))
-		}
-	}
 }
 
 func TestSimpleBroadcastTo(t *testing.T) {
@@ -286,6 +281,36 @@ func TestBroadcastCancelOneClientFails(t *testing.T) {
 	}
 }
 
+func TestBroadcastCallOrdering(t *testing.T) {
+	numSrvs := 3
+	numReqs := 10
+	_, srvAddrs, srvCleanup, err := createSrvs(numSrvs)
+	if err != nil {
+		t.Error(err)
+	}
+	defer srvCleanup()
+
+	config, clientCleanup, err := newClient(srvAddrs, "127.0.0.1:8080")
+	if err != nil {
+		t.Error(err)
+	}
+	defer clientCleanup()
+
+	for i := 1; i <= numReqs; i++ {
+		//slog.Info("req", "no", i)
+		val := int64(i * 100)
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		resp, err := config.Order(ctx, &Request{Value: val})
+		if err != nil {
+			t.Error(err)
+		}
+		if resp.GetResult() != 0 {
+			t.Error(fmt.Sprintf("resp is wrong, want: %v, got: %v", val, resp.GetResult()))
+		}
+		cancel()
+	}
+}
+
 func TestBroadcastCallRace(t *testing.T) {
 	_, srvAddrs, srvCleanup, err := createSrvs(3)
 	if err != nil {