Merge pull request #1034 from rqlite/queued-load-test

Add queued write load test

system_test/cluster_test.go

@@ -660,6 +660,100 @@ func Test_MultiNodeClusterNodes(t *testing.T) {
 	}
 }
 
+// Test_MultiNodeClusterQueuedWrites tests writing to a cluster using
+// normal and queued writes.
+func Test_MultiNodeClusterQueuedWrites(t *testing.T) {
+	node1 := mustNewLeaderNode()
+	defer node1.Deprovision()
+
+	if _, err := node1.Execute(`CREATE TABLE foo (id integer not null primary key, name text)`); err != nil {
+		t.Fatalf("failed to create table: %s", err.Error())
+	}
+
+	// Join a second and third nodes
+	node2 := mustNewNode(false)
+	defer node2.Deprovision()
+	if err := node2.Join(node1); err != nil {
+		t.Fatalf("node failed to join leader: %s", err.Error())
+	}
+	_, err := node2.WaitForLeader()
+	if err != nil {
+		t.Fatalf("failed waiting for leader: %s", err.Error())
+	}
+	node3 := mustNewNode(false)
+	defer node3.Deprovision()
+	if err := node3.Join(node1); err != nil {
+		t.Fatalf("node failed to join leader: %s", err.Error())
+	}
+	_, err = node3.WaitForLeader()
+	if err != nil {
+		t.Fatalf("failed waiting for leader: %s", err.Error())
+	}
+
+	// Write data to the cluster, via various methods and nodes.
+	var wg sync.WaitGroup
+	wg.Add(5)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 100; i++ {
+			if _, err := node1.Execute(`INSERT INTO foo(name) VALUES("fiona")`); err != nil {
+				t.Fatalf("failed to create table: %s", err.Error())
+			}
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 100; i++ {
+			if _, err := node2.Execute(`INSERT INTO foo(name) VALUES("fiona")`); err != nil {
+				t.Fatalf("failed to create table: %s", err.Error())
+			}
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 99; i++ {
+			if _, err := node2.ExecuteQueued(`INSERT INTO foo(name) VALUES("fiona")`, false); err != nil {
+				t.Fatalf("failed to create table: %s", err.Error())
+			}
+		}
+		if _, err := node2.ExecuteQueued(`INSERT INTO foo(name) VALUES("fiona")`, true); err != nil {
+			t.Fatalf("failed to create table: %s", err.Error())
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 99; i++ {
+			if _, err := node3.ExecuteQueued(`INSERT INTO foo(name) VALUES("fiona")`, false); err != nil {
+				t.Fatalf("failed to create table: %s", err.Error())
+			}
+		}
+		if _, err := node3.ExecuteQueued(`INSERT INTO foo(name) VALUES("fiona")`, true); err != nil {
+			t.Fatalf("failed to create table: %s", err.Error())
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 99; i++ {
+			if _, err := node3.ExecuteQueued(`INSERT INTO foo(name) VALUES("fiona")`, false); err != nil {
+				t.Fatalf("failed to create table: %s", err.Error())
+			}
+		}
+		if _, err := node3.ExecuteQueued(`INSERT INTO foo(name) VALUES("fiona")`, true); err != nil {
+			t.Fatalf("failed to create table: %s", err.Error())
+		}
+	}()
+	wg.Wait()
+
+	r, err := node1.Query(`SELECT COUNT(*) FROM foo`)
+	if err != nil {
+		t.Fatalf("failed to query follower node: %s", err.Error())
+	}
+
+	if got, exp := r, `{"results":[{"columns":["COUNT(*)"],"types":[""],"values":[[500]]}]}`; got != exp {
+		t.Fatalf("incorrect count, got %s, exp %s", got, exp)
+	}
+}
+
 // Test_MultiNodeClusterNodesNonVoter checks nodes/ endpoint with a non-voting node.
 func Test_MultiNodeClusterNodesNonVoter(t *testing.T) {
 	node1 := mustNewLeaderNode()