on emore

cmd/kv-write-canary/go.mod

@@ -0,0 +1,10 @@
+module kv-write-canary
+
+go 1.24.2
+
+require github.com/redis/go-redis/v9 v9.14.0
+
+require (
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+)

cmd/kv-write-canary/go.sum

@@ -0,0 +1,10 @@
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/redis/go-redis/v9 v9.14.0 h1:u4tNCjXOyzfgeLN+vAZaW1xUooqWDqVEsZN0U01jfAE=
+github.com/redis/go-redis/v9 v9.14.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=

cmd/kv-write-canary/main.go

@@ -0,0 +1,124 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	valkey "github.com/redis/go-redis/v9"
+)
+
+func main() {
+	url := os.Args[1]
+
+	ctx := context.Background()
+
+	opt, err := valkey.ParseURL(url)
+	if err != nil {
+		panic(err)
+	}
+
+	client := valkey.NewClient(opt)
+	if err != nil {
+		panic(err)
+	}
+	defer client.Close()
+
+	if err := client.Ping(ctx).Err(); err != nil {
+		panic(err)
+	}
+
+	spans := poll(ctx, client)
+	format := "15:04:05"
+	for i, span := range spans.Spans {
+		log.Printf("[%d] (%vs) %v .. %v", i, int(span[1].Sub(span[0]).Seconds()), span[0].Format(format), span[1].Format(format))
+	}
+
+	misses := 0
+	for i, write := range spans.Writes {
+		result := client.Get(ctx, write)
+		if err := result.Err(); err != nil {
+			misses += 1
+			log.Printf("[%d] expected %s but %v", i, write, err)
+		}
+	}
+	log.Printf("cannot confirm %d/%d entries", misses, len(spans.Writes))
+}
+
+type Poll struct {
+	Spans  [][2]time.Time
+	Writes []string
+}
+
+func poll(ctx context.Context, client *valkey.Client) Poll {
+	ctx, can := signal.NotifyContext(context.Background(), syscall.SIGINT)
+	defer can()
+
+	spans := [][2]time.Time{
+		{time.Now(), time.Now()},
+	}
+	writes := []string{}
+	for ctx.Err() == nil {
+		waitFailure(ctx, client, func(s string) {
+			writes = append(writes, s)
+		})
+		spans[len(spans)-1][1] = time.Now()
+		waitSuccess(ctx, client, func(s string) {
+			writes = append(writes, s)
+		})
+		spans = append(spans, [2]time.Time{time.Now(), time.Now()})
+	}
+	return Poll{Spans: spans, Writes: writes}
+}
+
+func waitFailure(ctx context.Context, client *valkey.Client, cb func(string)) {
+	waitResult(ctx, client, false, cb)
+}
+
+func waitSuccess(ctx context.Context, client *valkey.Client, cb func(string)) {
+	waitResult(ctx, client, true, cb)
+}
+
+func waitResult(ctx context.Context, client *valkey.Client, wantOK bool, cb func(string)) {
+	wait(ctx, func() bool {
+		k := time.Now().String()
+		result := client.Do(ctx, "SET", k, "ok")
+		err := result.Err()
+		got := err == nil
+		if got {
+			cb(k)
+		}
+		if got != wantOK {
+			fmt.Fprintf(os.Stderr, ".")
+		} else if wantOK {
+			fmt.Fprintf(os.Stderr, "\n")
+			log.Println("okay!")
+		} else {
+			fmt.Fprintf(os.Stderr, "\n")
+			log.Println("failed!")
+		}
+		return got == wantOK
+	})
+}
+
+func wait(ctx context.Context, foo func() bool) {
+	c := time.NewTicker(250 * time.Millisecond)
+	defer c.Stop()
+
+	for {
+		select {
+		case <-c.C:
+		case <-ctx.Done():
+			fmt.Println()
+			return
+		}
+
+		if foo() {
+			return
+		}
+	}
+}