valkey-go

The valkeycompat.Adapter provides a high-level API that closely mimics the go-redis Cmdable interface. This is useful for developers who prefer a more concise syntax than the standard valkey-go command builder or those migrating existing codebases from go-redis to valkey.

To migrate, replace your go-redis UniversalClient with a valkeycompat.Adapter initialized with a valkey.Client.

import (
	"github.com/valkey-io/valkey-go"
	"github.com/valkey-io/valkey-go/valkeycompat"
)

// ... setup client ...
compat := valkeycompat.NewAdapter(client)
// Use compat as a go-redis replacement

The valkeyaside package provides an implementation of the Cache-Aside pattern enhanced by Valkey Client-Side Caching. This approach addresses common issues like stale cache overrides and cache stampedes.

Key benefits include:

• Reduced Network Round Trips: Valkey proactively invalidates the client-side cache.
• Cache Stampede Prevention: Uses locking (similar to valkeylock) so only the first cache miss triggers a database update, while subsequent callers wait for notifications.

To prevent lost updates in a concurrent environment, add an int64 field to your struct with the valkey:",ver" tag.

When using this feature:

1. repo.Save(ctx, entity) will check if the version in Valkey matches the version in your struct.
2. If they do not match, Save() returns ErrVersionMismatch.
3. If they match, the save succeeds and the version in Valkey is automatically incremented.

If you do not include a field with the valkey:",ver" tag, Save() will always succeed without version checking.

type Example struct {
    Key string `json:"key" valkey:",key"` 
    Ver int64  `json:"ver" valkey:",ver"` // Enables optimistic locking
}

// If Ver is modified externally, this will fail:
exp.Ver = 0
err := repo.Save(ctx, exp)
if err == om.ErrVersionMismatch {
    // Handle conflict
}

By default, valkey-go uses Auto Pipelining for all concurrent non-blocking commands (like GET or SET). This automatically reduces round trips and system calls by grouping commands. You can leverage this simply by calling client.Do() from multiple goroutines concurrently.

func BenchmarkPipelining(b *testing.B, client valkey.Client) {
  b.RunParallel(func(pb *testing.PB) {
    for pb.Next() {
      client.Do(context.Background(), client.B().Get().Key("k").Build()).ToString()
    }
  })
}

The valkeylock implementation leverages Valkey Client-Side Caching to provide high-performance distributed locking:

1. Acquisition: It attempts to acquire a majority of keys (e.g., 2 out of 3 if KeyMajority is 2) using SET NX PXAT (or SET NX PX if FallbackSETPX is true).
2. Success: If a majority is reached within KeyValidity, WithContext returns. The Locker then periodically extends the lock and watches for client-side caching notifications.
3. Failure/Loss: If the majority is lost (nodes go down or keys are deleted), the Locker receives a notification and immediately cancels the returned ctx.
4. Waiting: If acquisition fails, the client waits for client-side caching notifications to trigger a retry, rather than polling.

The valkeylimiter module implements a Fixed Window Algorithm for distributed rate limiting. It uses Valkey to maintain counters, ensuring consistency across multiple distributed environments.

Key Mechanisms:

• Atomicity: It uses Lua scripts executed within Valkey to ensure that checking and updating rate limits are atomic operations, preventing race conditions.
• Automatic Resets: It leverages Valkey's expiration capabilities to automatically reset rate limits after the specified time window, which ensures efficient memory usage.
• Client Feedback: The limiter provides ResetAtMs timestamps, allowing your application to inform clients exactly when they can retry requests.

Methods like client.Do(), client.DoMulti(), client.DoCache(), and client.DoMultiCache() can return early if the provided context deadline is reached.

Important: Even if the method returns early due to a timeout, the command has likely already been sent to the server.

To ensure manual context cancellation is respected (especially for blocking requests on dedicated connections), you can use ClientOption.AlwaysPipelining to start pipeline mode in advance.

ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
client.Do(ctx, client.B().Set().Key("key").Value("val").Nx().Build()).Error() == context.DeadlineExceeded

You can automate record expiration by adding a time.Time field to your struct with the valkey:",exat" tag. When repo.Save(ctx, entity) is called, the repository will issue a PEXPIREAT command to set the expiry based on that timestamp.

Note: If the time.Time field is set to its zero value, the expiry of the record in Valkey remains unchanged.

type Example struct {
    Key  string    `json:"key" valkey:",key"` 
    ExAt time.Time `json:"exat" valkey:",exat"` 
}

exp := repo.NewEntity()
exp.ExAt = time.Now().Add(time.Hour) // Record expires in 1 hour
repo.Save(ctx, exp)