sql,metrics: do not increment ROLLBACK counter if in CommitWait

Release note (bug fix): Previously if `RELEASE SAVEPOINT
cockroach_restart` was followed by `ROLLBACK`, the
`sql.txn.rollback.count` metric would be incremented. This was
incorrect, since the txn had already committed. Now that metric is not
incremented in this case.

pkg/server/status_test.go

@@ -1160,6 +1160,44 @@ func TestStatusVars(t *testing.T) {
 	}
 }
 
+// TestStatusVarsTxnMetrics verifies that the metrics from the /_status/vars
+// endpoint for txns and the special cockroach_restart savepoint are correct.
+func TestStatusVarsTxnMetrics(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer db.Close()
+	defer s.Stopper().Stop(context.Background())
+
+	if _, err := db.Exec("BEGIN;" +
+		"SAVEPOINT cockroach_restart;" +
+		"SELECT 1;" +
+		"RELEASE SAVEPOINT cockroach_restart;" +
+		"ROLLBACK;"); err != nil {
+		t.Fatal(err)
+	}
+
+	body, err := getText(s, s.AdminURL()+statusPrefix+"vars")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(body, []byte("sql_txn_begin_count 1")) {
+		t.Errorf("expected `sql_txn_begin_count 1`, got: %s", body)
+	}
+	if !bytes.Contains(body, []byte("sql_restart_savepoint_count 1")) {
+		t.Errorf("expected `sql_restart_savepoint_count 1`, got: %s", body)
+	}
+	if !bytes.Contains(body, []byte("sql_restart_savepoint_release_count 1")) {
+		t.Errorf("expected `sql_restart_savepoint_release_count 1`, got: %s", body)
+	}
+	if !bytes.Contains(body, []byte("sql_txn_commit_count 1")) {
+		t.Errorf("expected `sql_txn_commit_count 1`, got: %s", body)
+	}
+	if !bytes.Contains(body, []byte("sql_txn_rollback_count 0")) {
+		t.Errorf("expected `sql_txn_rollback_count 0`, got: %s", body)
+	}
+}
+
 func TestSpanStatsResponse(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/sql/conn_executor.go

@@ -2701,7 +2701,13 @@ func (sc *StatementCounters) incrementCount(ex *connExecutor, stmt tree.Statemen
 	case *tree.CommitTransaction:
 		sc.TxnCommitCount.Inc()
 	case *tree.RollbackTransaction:
-		sc.TxnRollbackCount.Inc()
+		// The CommitWait state means that the transaction has already committed
+		// after a specially handled `RELEASE SAVEPOINT cockroach_restart` command.
+		if ex.getTransactionState() == CommitWaitStateStr {
+			sc.TxnCommitCount.Inc()
+		} else {
+			sc.TxnRollbackCount.Inc()
+		}
 	case *tree.Savepoint:
 		if ex.isCommitOnReleaseSavepoint(t.Name) {
 			sc.RestartSavepointCount.Inc()