Check that queue and transactions work

http/service.go

@@ -1358,14 +1358,16 @@ func (s *Service) runQueue() {
 					if err == store.ErrNotLeader {
 						addr, err := s.store.LeaderAddr()
 						if err != nil || addr == "" {
-							s.logger.Println("execute queue can't find leader")
+							s.logger.Println("execute queue can't find leader for sequence number %d",
+								req.SequenceNumber)
 							stats.Add(numQueuedExecutionsFailed, 1)
 							time.Sleep(retryDelay)
 							continue
 						}
 						_, err = s.cluster.Execute(er, addr, defaultTimeout)
 						if err != nil {
-							s.logger.Printf("execute queue write failed: %s", err.Error())
+							s.logger.Printf("execute queue write failed for sequence number %d: %s",
+								req.SequenceNumber, err.Error())
 							time.Sleep(retryDelay)
 							continue
 						}

system_test/single_node_test.go

@@ -389,6 +389,83 @@ LOOP:
 	}
 }
 
+// Test_SingleNodeQueuedBadStmt tests that a single bad SQL statement has the right outcome.
+func Test_SingleNodeQueuedBadStmt(t *testing.T) {
+	node := mustNewLeaderNode()
+	defer node.Deprovision()
+	node.Service.DefaultQueueTx = false
+
+	ticker := time.NewTicker(10 * time.Millisecond)
+	defer ticker.Stop()
+
+	_, err := node.Execute(`CREATE TABLE foo (id integer not null primary key, name text)`)
+	if err != nil {
+		t.Fatalf(`CREATE TABLE failed: %s`, err.Error())
+	}
+
+	qWrites := []string{
+		`INSERT INTO foo(name) VALUES("fiona")`,
+		`INSERT INTO nonsense`,
+		`INSERT INTO foo(name) VALUES("fiona")`,
+	}
+	resp, err := node.ExecuteQueuedMulti(qWrites, false)
+	if err != nil {
+		t.Fatalf(`queued write failed: %s`, err.Error())
+	}
+	if !QueuedResponseRegex.MatchString(resp) {
+		t.Fatalf("queued response is not valid: %s", resp)
+	}
+
+	timer := time.NewTimer(5 * time.Second)
+	defer timer.Stop()
+
+LOOP1:
+	for {
+		select {
+		case <-ticker.C:
+			r, err := node.Query(`SELECT COUNT(*) FROM foo`)
+			if err != nil {
+				t.Fatalf(`query failed: %s`, err.Error())
+			}
+			if r == `{"results":[{"columns":["COUNT(*)"],"types":[""],"values":[[2]]}]}` {
+				break LOOP1
+			}
+		case <-timer.C:
+			t.Fatalf("timed out waiting for queued writes")
+
+		}
+	}
+
+	// Enable transactions, so that the next request shouldn't change the data
+	// due to the single bad statement.
+	node.Service.DefaultQueueTx = true
+	resp, err = node.ExecuteQueuedMulti(qWrites, false)
+	if err != nil {
+		t.Fatalf(`queued write failed: %s`, err.Error())
+	}
+	if !QueuedResponseRegex.MatchString(resp) {
+		t.Fatalf("queued response is not valid: %s", resp)
+	}
+
+	timer.Reset(5 * time.Second)
+LOOP2:
+	for {
+		select {
+		case <-ticker.C:
+			r, err := node.Query(`SELECT COUNT(*) FROM foo`)
+			if err != nil {
+				t.Fatalf(`query failed: %s`, err.Error())
+			}
+			if r == `{"results":[{"columns":["COUNT(*)"],"types":[""],"values":[[2]]}]}` {
+				break LOOP2
+			}
+		case <-timer.C:
+			t.Fatalf("timed out waiting for queued writes")
+
+		}
+	}
+}
+
 // Test_SingleNodeSQLInjection demonstrates that using the non-parameterized API is vulnerable to
 // SQL injection attacks.
 func Test_SingleNodeSQLInjection(t *testing.T) {