redis: leader-route demo reads and fail invalid Jepsen runs

Author: bootjp

adapter/add_voter_join_test.go

@@ -182,7 +182,7 @@ func startAddVoterJoinNode(
 		return err
 	})
 
-	rd := NewRedisServer(lis.redis, port.redisAddress, st, coordinator, leaderRedisMap, relay)
+	rd := NewRedisServer(lis.redis, port.redisAddress, routedStore, coordinator, leaderRedisMap, relay)
 	workers.Go(func() error {
 		err := rd.Run()
 		if errors.Is(err, net.ErrClosed) {

adapter/redis_server_setup_test.go

@@ -0,0 +1,19 @@
+package adapter
+
+import (
+	"testing"
+
+	"github.com/bootjp/elastickv/kv"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRedisServerUsesLeaderRoutedStoreInSingleGroupCluster(t *testing.T) {
+	t.Parallel()
+
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+
+	require.IsType(t, &kv.LeaderRoutedStore{}, nodes[0].redisServer.store)
+	require.IsType(t, &kv.LeaderRoutedStore{}, nodes[1].redisServer.store)
+	require.IsType(t, &kv.LeaderRoutedStore{}, nodes[2].redisServer.store)
+}

adapter/test_util.go

@@ -370,7 +370,7 @@ func setupNodes(t *testing.T, ctx context.Context, n int, ports []portsAdress) (
 			assert.NoError(t, srv.Serve(lis))
 		}(s, grpcSock)
 
-		rd := NewRedisServer(redisSock, port.redisAddress, st, coordinator, leaderRedisMap, relay)
+		rd := NewRedisServer(redisSock, port.redisAddress, routedStore, coordinator, leaderRedisMap, relay)
 		go func(server *RedisServer) {
 			assert.NoError(t, server.Run())
 		}(rd)

cmd/server/demo.go

@@ -354,7 +354,8 @@ func setupRedis(ctx context.Context, lc net.ListenConfig, st store.MVCCStore, co
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
-	return adapter.NewRedisServer(l, redisAddr, st, coordinator, leaderRedis, relay), nil
+	routedStore := kv.NewLeaderRoutedStore(st, coordinator)
+	return adapter.NewRedisServer(l, redisAddr, routedStore, coordinator, leaderRedis, relay), nil
 }
 
 func run(ctx context.Context, eg *errgroup.Group, cfg config) error {

jepsen/src/elastickv/redis_workload.clj

@@ -175,6 +175,13 @@
     :parse-fn #(Integer/parseInt %)]
    ["-h" "--help"]])
 
+(defn fail-on-invalid!
+  "Raises when Jepsen completed analysis and found the history invalid."
+  [result]
+  (when (false? (:valid? result))
+    (throw (ex-info "Jepsen analysis invalid" {:result result})))
+  result)
+
 (defn -main
   [& args]
   (let [{:keys [options errors summary]} (tools.cli/parse-opts args cli-opts)
@@ -211,5 +218,5 @@
       (seq errors) (binding [*out* *err*]
                      (println "Error parsing options:" (str/join "; " errors)))
       (:local options) (binding [control/*dummy* true]
-                         (jepsen/run! (elastickv-redis-test options)))
-      :else (jepsen/run! (elastickv-redis-test options)))))
+                         (fail-on-invalid! (jepsen/run! (elastickv-redis-test options))))
+      :else (fail-on-invalid! (jepsen/run! (elastickv-redis-test options))))))

jepsen/test/elastickv/redis_workload_test.clj

@@ -0,0 +1,13 @@
+(ns elastickv.redis-workload-test
+  (:require [clojure.test :refer :all]
+            [elastickv.redis-workload :as workload]))
+
+(deftest fail-on-invalid-passes-through-valid-results
+  (let [result {:valid? true}]
+    (is (= result (workload/fail-on-invalid! result)))))
+
+(deftest fail-on-invalid-throws-for-invalid-results
+  (is (thrown-with-msg?
+       clojure.lang.ExceptionInfo
+       #"Jepsen analysis invalid"
+       (workload/fail-on-invalid! {:valid? false}))))