test(19.3-05): cooler caches, rate-limit parity, Install and after-commit purge
- miniredis v2.40.0 as a test-only dependency (approved at the Task 4 check) - memory and Redis cache suites, Flush limited to <prefix>cache: - one parity table through surf.MemoryStore, miniredis and real Redis - Install, FromApp and site_cache validation; Postgres after-commit purge - gate: cooler names in the go and docker stages
This commit is contained in:
2
go.mod
2
go.mod
@@ -7,6 +7,7 @@ ignore ./admin/node_modules
|
||||
|
||||
require (
|
||||
github.com/alecthomas/chroma/v2 v2.27.0
|
||||
github.com/alicebob/miniredis/v2 v2.40.0
|
||||
github.com/disintegration/imaging v1.6.2
|
||||
github.com/fsnotify/fsnotify v1.10.1
|
||||
github.com/go-gormigrate/gormigrate/v2 v2.1.7
|
||||
@@ -128,6 +129,7 @@ require (
|
||||
github.com/tidwall/sjson v1.2.5 // indirect
|
||||
github.com/tklauser/go-sysconf v0.4.0 // indirect
|
||||
github.com/tklauser/numcpus v0.12.0 // indirect
|
||||
github.com/yuin/gopher-lua v1.1.1 // indirect
|
||||
github.com/yusufpapurcu/wmi v1.2.4 // indirect
|
||||
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
|
||||
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect
|
||||
|
||||
4
go.sum
4
go.sum
@@ -36,6 +36,8 @@ github.com/alecthomas/chroma/v2 v2.27.0 h1:FodwmyOBgJULFYmDqibcp9pvfDLWdtPRh9v/r
|
||||
github.com/alecthomas/chroma/v2 v2.27.0/go.mod h1:NjJ3ciIgrqBNeIkWZ4e46nseoLDslxU1LmfCoL+wcY8=
|
||||
github.com/alecthomas/repr v0.5.2 h1:SU73FTI9D1P5UNtvseffFSGmdNci/O6RsqzeXJtP0Qs=
|
||||
github.com/alecthomas/repr v0.5.2/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4=
|
||||
github.com/alicebob/miniredis/v2 v2.40.0 h1:6IgTQbvm2JiMYZ8yur/acNJzVjtSOWCelxXbk/LUBXc=
|
||||
github.com/alicebob/miniredis/v2 v2.40.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
|
||||
github.com/aws/aws-sdk-go-v2 v1.41.9 h1:/rYeyO2+HrMztAmxAq9++XJtFMqSIpSsNA0yDGALYq4=
|
||||
github.com/aws/aws-sdk-go-v2 v1.41.9/go.mod h1:+HsoOEX80qAVUitj1A2DhCNTjmb3edVyuDypb6LNEeo=
|
||||
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.11 h1:h5+3VT69KUBK24grGuuA5saDJTj2IIjLb9au668Fo5I=
|
||||
@@ -311,6 +313,8 @@ github.com/wneessen/go-mail v0.8.1 h1:tVcncj02/QySVFw3zr/kXOzZcuFQqBNT6K+Rbgm/pc
|
||||
github.com/wneessen/go-mail v0.8.1/go.mod h1:dWZ61zadzCIyvB4y1/YzC5O7MrbbzBfPkARmbosdf8w=
|
||||
github.com/yuin/goldmark v1.8.6 h1:d0VcaP1sx9GkFVkoW+KtggpGi2KZ965i14b0+bDQST4=
|
||||
github.com/yuin/goldmark v1.8.6/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg=
|
||||
github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
|
||||
github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
|
||||
github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
|
||||
github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
|
||||
github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
|
||||
|
||||
@@ -212,12 +212,15 @@ func AllowExport(ctx context.Context, app *backpack.App, fallback cooler.Attempt
|
||||
- Third-party: `github.com/redis/go-redis/v9` (Redis client), `github.com/maypok86/otter/v2` (memory driver).
|
||||
- Third-party (purge): `gorm.io/gorm` (the save transaction handle).
|
||||
- Standard library: `bytes`, `context`, `encoding/json`, `errors`, `fmt`, `io`, `log/slog`, `net`, `net/http`, `net/url`, `strconv`, `strings`, `sync`, `time`.
|
||||
- Tests additionally use `github.com/testcontainers/testcontainers-go` (Redis container).
|
||||
- Tests additionally use `github.com/alicebob/miniredis/v2` (an in-process Redis for the `-short` suites; only `_test.go` files import it) and `github.com/testcontainers/testcontainers-go` (Redis and Postgres containers).
|
||||
|
||||
## Testing
|
||||
|
||||
```sh
|
||||
go test ./modules/cooler/...
|
||||
go test -short ./modules/cooler/... # no Docker: memory driver, miniredis, httptest receivers
|
||||
go test ./modules/cooler/... # also the Redis and Postgres containers (needs Docker)
|
||||
```
|
||||
|
||||
`TestSiteCachePurgeConfig` and `TestSiteCachePurgeRequest` cover the purge configuration and the request against an `httptest` receiver without external services. `TestRedisCacheSmoke` starts a Redis container through testcontainers-go and needs Docker; `go test -short ./modules/cooler/...` skips it, and the other tests need no external services. The cross-instance rate-limit test lives in surf (`go test ./modules/surf -run TestRateLimitAcrossInstancesSmoke`), starts a Redis container through testcontainers-go and needs Docker; `-short` skips it.
|
||||
The `-short` suites need no external services. `TestMemoryCache`, `TestMemoryCacheBounded` and `TestMemoryIncrementConcurrent` cover the memory driver; `TestRedisCache` and `TestRedisFlushOnlyPrefix` run the Redis driver on miniredis (TTLs through `FastForward`). `TestRateLimitParity` runs one table of attempts through `surf.NewMemoryStore` and the Redis store and expects the same answers; `TestRateLimitFallbackOnError` stops Redis and checks the fallback store answers with one warning. `TestInstallConfigErrors`, `TestInstallPingWarning` and `TestFromAppLazy` cover start-up; `TestSiteCachePurgeConfig` and `TestSiteCachePurgeRequest` the purge configuration and request against an `httptest` receiver.
|
||||
|
||||
The Docker suites skip under `-short`: `TestRateLimitParityDocker` and `TestRedisCacheDocker` run the same checks on a `redis:7-alpine` container, `TestSiteCacheAfterCommitOnly` and `TestSiteCacheFailureNeverFails` run the after-commit purge in Postgres transactions, and `TestRedisCacheSmoke` is the original Redis smoke test. The cross-instance rate-limit test lives in surf (`go test ./modules/surf -run TestRateLimitAcrossInstancesSmoke`).
|
||||
|
||||
133
modules/cooler/docker_test.go
Normal file
133
modules/cooler/docker_test.go
Normal file
@@ -0,0 +1,133 @@
|
||||
package cooler_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/cooler"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/testcontainers/testcontainers-go"
|
||||
"github.com/testcontainers/testcontainers-go/wait"
|
||||
)
|
||||
|
||||
// realRedis starts redis:7-alpine for the test and returns a client from
|
||||
// OpenRedis. It skips under -short.
|
||||
func realRedis(t *testing.T) *redis.Client {
|
||||
t.Helper()
|
||||
if testing.Short() {
|
||||
t.Skip("requires testcontainers redis")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
t.Cleanup(cancel)
|
||||
ctr, err := testcontainers.Run(ctx, "redis:7-alpine",
|
||||
testcontainers.WithExposedPorts("6379/tcp"),
|
||||
testcontainers.WithWaitStrategy(wait.ForListeningPort("6379/tcp")),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("redis: testcontainers: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = testcontainers.TerminateContainer(ctr) })
|
||||
host, err := ctr.Host(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
port, err := ctr.MappedPort(ctx, "6379/tcp")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
client, err := cooler.OpenRedis(ctx, fmt.Sprintf("redis://%s:%s/0", host, port.Port()))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = client.Close() })
|
||||
if err := client.Ping(ctx).Err(); err != nil {
|
||||
t.Fatalf("redis ping: %v", err)
|
||||
}
|
||||
return client
|
||||
}
|
||||
|
||||
func TestRateLimitParityDocker(t *testing.T) {
|
||||
client := realRedis(t)
|
||||
store := cooler.NewRateLimitStore(client, "acme:", nil)
|
||||
runParity(t, parityStore{name: "redis", store: store, advance: sleepFor})
|
||||
|
||||
if got := countAdmitted(cooler.NewRateLimitStore(client, "burst:", nil), 50, 10); got != 10 {
|
||||
t.Fatalf("real Redis admitted %d of 50, want 10", got)
|
||||
}
|
||||
ctx := t.Context()
|
||||
if v, err := client.Get(ctx, "burst:rl:burst").Result(); err != nil || v != "10" {
|
||||
t.Fatalf("burst counter = %q %v, want 10", v, err)
|
||||
}
|
||||
if ttl, err := client.PTTL(ctx, "burst:rl:burst").Result(); err != nil || ttl <= 0 || ttl > time.Minute {
|
||||
t.Fatalf("burst window TTL = %s %v", ttl, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRedisCacheDocker(t *testing.T) {
|
||||
client := realRedis(t)
|
||||
ctx := t.Context()
|
||||
c := cooler.NewRedisCache(client, "acme:")
|
||||
|
||||
if err := c.Set(ctx, "greeting", []byte("hello"), 150*time.Millisecond); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if v, ok, err := c.Get(ctx, "greeting"); err != nil || !ok || string(v) != "hello" {
|
||||
t.Fatalf("Get = %q %v %v", v, ok, err)
|
||||
}
|
||||
if ttl, _ := client.PTTL(ctx, "acme:cache:greeting").Result(); ttl <= 0 || ttl > 150*time.Millisecond {
|
||||
t.Fatalf("PTTL = %s, want (0, 150ms]", ttl)
|
||||
}
|
||||
time.Sleep(250 * time.Millisecond)
|
||||
if ok, _ := c.Has(ctx, "greeting"); ok {
|
||||
t.Fatal("the key outlived its TTL")
|
||||
}
|
||||
|
||||
if err := c.Set(ctx, "views", []byte("1"), time.Minute); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n, err := c.Increment(ctx, "views", 2); err != nil || n != 3 {
|
||||
t.Fatalf("Increment = %d %v", n, err)
|
||||
}
|
||||
if ttl, _ := client.PTTL(ctx, "acme:cache:views").Result(); ttl <= 50*time.Second {
|
||||
t.Fatalf("Increment dropped the TTL: %s", ttl)
|
||||
}
|
||||
calls := 0
|
||||
for range 2 {
|
||||
v, err := c.Remember(ctx, "remembered", time.Minute, func(context.Context) ([]byte, error) { calls++; return []byte("computed"), nil })
|
||||
if err != nil || string(v) != "computed" {
|
||||
t.Fatalf("Remember = %q %v", v, err)
|
||||
}
|
||||
}
|
||||
if calls != 1 {
|
||||
t.Fatalf("Remember called the callback %d times", calls)
|
||||
}
|
||||
|
||||
// Flush walks SCAN in batches: more keys than one batch, and the keys
|
||||
// outside <prefix>cache: survive.
|
||||
pipe := client.Pipeline()
|
||||
for i := range 1200 {
|
||||
pipe.Set(ctx, "acme:cache:bulk"+strconv.Itoa(i), "v", 0)
|
||||
}
|
||||
survivors := []string{"acme:rl:login|192.0.2.1", "other:cache:greeting", "acme:cachex"}
|
||||
for _, k := range survivors {
|
||||
pipe.Set(ctx, k, "keep", 0)
|
||||
}
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := c.Flush(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
left, err := client.Keys(ctx, "acme:cache:*").Result()
|
||||
if err != nil || len(left) != 0 {
|
||||
t.Fatalf("Flush left %d cache keys (%v)", len(left), err)
|
||||
}
|
||||
for _, k := range survivors {
|
||||
if n, err := client.Exists(ctx, k).Result(); err != nil || n != 1 {
|
||||
t.Fatalf("Flush removed %q", k)
|
||||
}
|
||||
}
|
||||
}
|
||||
304
modules/cooler/install_test.go
Normal file
304
modules/cooler/install_test.go
Normal file
@@ -0,0 +1,304 @@
|
||||
package cooler_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
"git.golem15.com/golem15/summercms/modules/cooler"
|
||||
"git.golem15.com/golem15/summercms/modules/surf"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
)
|
||||
|
||||
// appWithLogger returns an app on cfg whose published logger writes to the
|
||||
// returned buffer.
|
||||
func appWithLogger(t *testing.T, cfg map[string]any) (*backpack.App, *lockedBuffer) {
|
||||
t.Helper()
|
||||
c := emptyConfig(t)
|
||||
for k, v := range cfg {
|
||||
mustSet(t, c, k, v)
|
||||
}
|
||||
app := backpack.New(c)
|
||||
buf := &lockedBuffer{}
|
||||
if err := app.Publish(slog.New(slog.NewTextHandler(buf, nil))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return app, buf
|
||||
}
|
||||
|
||||
func TestInstallConfigErrors(t *testing.T) {
|
||||
const secretURL = "redis://alice:s3cretpw@cache.example.test:6379/0\x7f"
|
||||
cases := []struct {
|
||||
name string
|
||||
cfg map[string]any
|
||||
names []string // substrings the error must contain
|
||||
}{
|
||||
{"unknown cache driver", map[string]any{cooler.CacheDriverKey: "memcached"}, []string{cooler.CacheDriverKey, "memcached"}},
|
||||
{"unknown rate-limit store", map[string]any{cooler.RateLimitStoreKey: "database"}, []string{cooler.RateLimitStoreKey, "database"}},
|
||||
{"redis store without redis.url", map[string]any{cooler.RateLimitStoreKey: "redis"}, []string{cooler.RedisURLKey, cooler.RateLimitStoreKey}},
|
||||
{"redis cache without redis.url", map[string]any{cooler.CacheDriverKey: "redis"}, []string{cooler.RedisURLKey, cooler.CacheDriverKey}},
|
||||
{"a malformed redis.url", map[string]any{cooler.RateLimitStoreKey: "redis", cooler.RedisURLKey: secretURL}, []string{cooler.RedisURLKey}},
|
||||
{"an unsupported scheme", map[string]any{cooler.CacheDriverKey: "redis", cooler.RedisURLKey: "http://cache.example.test"}, []string{cooler.RedisURLKey}},
|
||||
{"max_entries 0", map[string]any{cooler.CacheMaxEntriesKey: 0}, []string{cooler.CacheMaxEntriesKey}},
|
||||
{"max_entries not a number", map[string]any{cooler.CacheMaxEntriesKey: "lots"}, []string{cooler.CacheMaxEntriesKey}},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
app, _ := appWithLogger(t, tc.cfg)
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err == nil {
|
||||
_ = h.Close()
|
||||
t.Fatal("Install succeeded, want a boot error")
|
||||
}
|
||||
for _, want := range tc.names {
|
||||
if !strings.Contains(err.Error(), want) {
|
||||
t.Fatalf("error %q does not name %q", err, want)
|
||||
}
|
||||
}
|
||||
if strings.Contains(err.Error(), "alice") || strings.Contains(err.Error(), "s3cretpw") {
|
||||
t.Fatalf("error %q leaks the URL's credentials", err)
|
||||
}
|
||||
if c, ok := app.Lookup[cooler.Cache](); ok && c != nil {
|
||||
t.Fatal("a failed Install published a Cache")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
if _, err := cooler.Install(context.Background(), nil); err == nil {
|
||||
t.Fatal("Install(nil app) must fail")
|
||||
}
|
||||
if _, err := cooler.FromApp(nil); err == nil {
|
||||
t.Fatal("FromApp(nil app) must fail")
|
||||
}
|
||||
app, _ := appWithLogger(t, map[string]any{cooler.CacheDriverKey: "file"})
|
||||
if _, err := cooler.FromApp(app); err == nil || !strings.Contains(err.Error(), cooler.CacheDriverKey) {
|
||||
t.Fatalf("FromApp with an unknown driver = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInstallPingWarning(t *testing.T) {
|
||||
addr := deadAddr(t)
|
||||
app, logs := appWithLogger(t, map[string]any{
|
||||
cooler.RateLimitStoreKey: "redis",
|
||||
cooler.RedisURLKey: "redis://alice:s3cretpw@" + addr + "/0",
|
||||
})
|
||||
start := time.Now()
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err != nil {
|
||||
t.Fatalf("an unreachable Redis must not fail Install: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = h.Close() })
|
||||
if elapsed := time.Since(start); elapsed > cooler.PingTimeout+time.Second {
|
||||
t.Fatalf("Install took %s, want at most about %s", elapsed, cooler.PingTimeout)
|
||||
}
|
||||
lines := logs.lines()
|
||||
if len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], "endpoint="+addr) {
|
||||
t.Fatalf("log = %q, want one warning naming the endpoint", lines)
|
||||
}
|
||||
if strings.Contains(lines[0], "alice") || strings.Contains(lines[0], "s3cretpw") || strings.Contains(lines[0], "redis://") {
|
||||
t.Fatalf("warning leaks the URL: %s", lines[0])
|
||||
}
|
||||
if !h.RateLimitRedis() {
|
||||
t.Fatal("the Redis store must stay selected so it recovers when Redis answers")
|
||||
}
|
||||
// The cache stays the memory default.
|
||||
if v, err := h.Cache().Remember(context.Background(), "k", time.Minute, func(context.Context) ([]byte, error) { return []byte("v"), nil }); err != nil || string(v) != "v" {
|
||||
t.Fatalf("Remember = %q %v", v, err)
|
||||
}
|
||||
if err := h.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
if err := h.Close(); err != nil {
|
||||
t.Fatalf("a second Close: %v", err)
|
||||
}
|
||||
if h.RateLimitRedis() {
|
||||
t.Fatal("RateLimitRedis after Close")
|
||||
}
|
||||
var nilHandle *cooler.Handle
|
||||
if nilHandle.Close() != nil || nilHandle.Cache() != nil || nilHandle.RateLimitRedis() {
|
||||
t.Fatal("a nil Handle must be inert")
|
||||
}
|
||||
}
|
||||
|
||||
func TestInstallReusesFromApp(t *testing.T) {
|
||||
t.Run("memory: FromApp's Cache survives Install", func(t *testing.T) {
|
||||
app, _ := appWithLogger(t, nil)
|
||||
first, err := cooler.FromApp(app)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = h.Close() })
|
||||
if h.Cache() != first {
|
||||
t.Fatal("Install replaced the Cache FromApp published")
|
||||
}
|
||||
if again, _ := cooler.FromApp(app); again != first {
|
||||
t.Fatal("FromApp built a second Cache")
|
||||
}
|
||||
// With the memory store, RateLimitStore hands back the fallback itself.
|
||||
fallback := surf.NewMemoryStore(0)
|
||||
if h.RateLimitRedis() || h.RateLimitStore(fallback) != cooler.Attempter(fallback) {
|
||||
t.Fatal("the memory store must return the fallback unchanged")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("redis: Install publishes one Redis cache that FromApp returns", func(t *testing.T) {
|
||||
mr := miniredis.RunT(t)
|
||||
app, _ := appWithLogger(t, map[string]any{
|
||||
cooler.CacheDriverKey: "redis",
|
||||
cooler.CachePrefixKey: "acme:",
|
||||
cooler.RedisURLKey: "redis://" + mr.Addr() + "/0",
|
||||
})
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = h.Close() })
|
||||
c, err := cooler.FromApp(app)
|
||||
if err != nil || c != h.Cache() {
|
||||
t.Fatalf("FromApp = %v %v, want Install's Cache", c, err)
|
||||
}
|
||||
if err := c.Set(context.Background(), "k", []byte("v"), 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !mr.Exists("acme:cache:k") {
|
||||
t.Fatalf("keys = %q, want acme:cache:k", mr.Keys())
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("concurrent FromApp calls share one Cache", func(t *testing.T) {
|
||||
app, _ := appWithLogger(t, nil)
|
||||
var wg sync.WaitGroup
|
||||
got := make([]cooler.Cache, 20)
|
||||
for i := range got {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
got[i], _ = cooler.FromApp(app)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
for i, c := range got {
|
||||
if c == nil || c != got[0] {
|
||||
t.Fatalf("call %d returned %v, want the shared Cache", i, c)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestFromAppLazy(t *testing.T) {
|
||||
mr := miniredis.RunT(t)
|
||||
app, _ := appWithLogger(t, map[string]any{
|
||||
cooler.CacheDriverKey: "redis",
|
||||
cooler.RedisURLKey: "redis://" + mr.Addr() + "/0",
|
||||
})
|
||||
c, err := cooler.FromApp(app)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n := mr.TotalConnectionCount(); n != 0 {
|
||||
t.Fatalf("FromApp opened %d connections, want none before the first command", n)
|
||||
}
|
||||
if err := c.Set(context.Background(), "k", []byte("v"), 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n := mr.TotalConnectionCount(); n == 0 {
|
||||
t.Fatal("the first command did not dial")
|
||||
}
|
||||
|
||||
// A Redis that does not exist still builds at once.
|
||||
dead, _ := appWithLogger(t, map[string]any{
|
||||
cooler.CacheDriverKey: "redis",
|
||||
cooler.RedisURLKey: "redis://" + deadAddr(t) + "/0",
|
||||
})
|
||||
start := time.Now()
|
||||
if _, err := cooler.FromApp(dead); err != nil {
|
||||
t.Fatalf("FromApp with an unreachable Redis: %v", err)
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
|
||||
t.Fatalf("FromApp took %s; it must not dial", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInstallSiteCacheValidation(t *testing.T) {
|
||||
const token = "install-site-cache-token-0123456789abcd"
|
||||
t.Run("a bad site_cache configuration is a boot error", func(t *testing.T) {
|
||||
for _, cfg := range []map[string]any{
|
||||
{cooler.SiteCachePurgeURLKey: "http://127.0.0.1:3000/purge"},
|
||||
{cooler.SiteCachePurgeTokenKey: token},
|
||||
{cooler.SiteCachePurgeURLKey: "http://example.test/purge", cooler.SiteCachePurgeTokenKey: token},
|
||||
{cooler.SiteCachePurgeURLKey: "http://127.0.0.1/purge", cooler.SiteCachePurgeTokenKey: "short"},
|
||||
} {
|
||||
app, _ := appWithLogger(t, cfg)
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err == nil {
|
||||
_ = h.Close()
|
||||
t.Fatalf("Install(%v) succeeded, want a boot error", cfg)
|
||||
}
|
||||
if !strings.Contains(err.Error(), "site_cache.") || strings.Contains(err.Error(), token) {
|
||||
t.Fatalf("error %q must name a site_cache key and never the token", err)
|
||||
}
|
||||
if p, ok := app.Lookup[*cooler.SiteCachePurger](); ok && p != nil {
|
||||
t.Fatal("a failed Install published a purger")
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a valid configuration is published and logged by endpoint", func(t *testing.T) {
|
||||
app, logs := appWithLogger(t, map[string]any{
|
||||
cooler.SiteCachePurgeURLKey: "http://127.0.0.1:3000/api/__site__/cache/purge",
|
||||
cooler.SiteCachePurgeTokenKey: token,
|
||||
})
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = h.Close() })
|
||||
p := cooler.SiteCacheFromApp(app)
|
||||
if p == nil || p.Endpoint() != "127.0.0.1:3000/api/__site__/cache/purge" {
|
||||
t.Fatalf("published purger = %v", p)
|
||||
}
|
||||
lines := logs.lines()
|
||||
if len(lines) != 1 || !strings.Contains(lines[0], "site cache purge enabled") || strings.Contains(lines[0], token) {
|
||||
t.Fatalf("log = %q", lines)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a published alias purger is reused, even with bad framework keys", func(t *testing.T) {
|
||||
app, _ := appWithLogger(t, map[string]any{cooler.SiteCachePurgeURLKey: "ftp://nowhere"})
|
||||
alias, err := cooler.NewSiteCachePurgerFor("acme.purge_url", "acme.purge_token", "http://127.0.0.1:4000/purge", token)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := app.Publish(alias); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err != nil {
|
||||
t.Fatalf("Install with a published alias purger: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = h.Close() })
|
||||
if got := cooler.SiteCacheFromApp(app); got != alias {
|
||||
t.Fatal("Install replaced the alias purger")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("no keys: no purger", func(t *testing.T) {
|
||||
app, _ := appWithLogger(t, nil)
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = h.Close() })
|
||||
if p := cooler.SiteCacheFromApp(app); p != nil {
|
||||
t.Fatalf("SiteCacheFromApp = %v, want nil", p)
|
||||
}
|
||||
})
|
||||
}
|
||||
336
modules/cooler/memory_test.go
Normal file
336
modules/cooler/memory_test.go
Normal file
@@ -0,0 +1,336 @@
|
||||
package cooler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func newMemory(t *testing.T, prefix string, maxEntries int) *memoryCache {
|
||||
t.Helper()
|
||||
c, err := NewMemoryCache(prefix, maxEntries)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, ok := c.(*memoryCache)
|
||||
if !ok {
|
||||
t.Fatalf("NewMemoryCache returned %T", c)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
func TestMemoryCache(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
m := newMemory(t, "acme:", DefaultMaxEntries)
|
||||
|
||||
if v, ok, err := m.Get(ctx, "missing"); v != nil || ok || err != nil {
|
||||
t.Fatalf("Get(missing) = %q %v %v", v, ok, err)
|
||||
}
|
||||
if ok, err := m.Has(ctx, "missing"); ok || err != nil {
|
||||
t.Fatalf("Has(missing) = %v %v", ok, err)
|
||||
}
|
||||
if err := m.Forget(ctx, "missing"); err != nil {
|
||||
t.Fatalf("Forget(missing) = %v", err)
|
||||
}
|
||||
|
||||
// Keys live under <prefix>cache:<key>.
|
||||
if err := m.Set(ctx, "greeting", []byte("hello"), time.Minute); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := m.c.GetIfPresent("acme:cache:greeting"); !ok {
|
||||
t.Fatal("Set did not store acme:cache:greeting")
|
||||
}
|
||||
if v, ok, err := m.Get(ctx, "greeting"); err != nil || !ok || string(v) != "hello" {
|
||||
t.Fatalf("Get = %q %v %v", v, ok, err)
|
||||
}
|
||||
|
||||
// ttl 0 and a negative ttl keep the value without expiry.
|
||||
for _, ttl := range []time.Duration{0, -time.Second} {
|
||||
key := "forever" + ttl.String()
|
||||
if err := m.Set(ctx, key, []byte("x"), ttl); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
e, ok := m.c.GetIfPresent(m.key(key))
|
||||
if !ok || !e.expiresAt.IsZero() {
|
||||
t.Fatalf("ttl %s: entry %+v found=%v, want no expiry", ttl, e, ok)
|
||||
}
|
||||
}
|
||||
|
||||
// A nil value is stored and read back as found.
|
||||
if err := m.Set(ctx, "nil", nil, 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok, err := m.Get(ctx, "nil"); !ok || err != nil {
|
||||
t.Fatalf("Get(nil value) found=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
// Copy-on-set and copy-on-get.
|
||||
in := []byte("abc")
|
||||
_ = m.Set(ctx, "copy", in, 0)
|
||||
in[0] = 'X'
|
||||
out, _, _ := m.Get(ctx, "copy")
|
||||
out[1] = 'Y'
|
||||
if again, _, _ := m.Get(ctx, "copy"); string(again) != "abc" {
|
||||
t.Fatalf("stored value mutated: %q", again)
|
||||
}
|
||||
|
||||
// Increment: missing starts at 0, existing integer grows, a non-integer fails.
|
||||
if n, err := m.Increment(ctx, "hits", 5); err != nil || n != 5 {
|
||||
t.Fatalf("Increment(missing) = %d %v", n, err)
|
||||
}
|
||||
if n, err := m.Increment(ctx, "hits", -2); err != nil || n != 3 {
|
||||
t.Fatalf("Increment(-2) = %d %v", n, err)
|
||||
}
|
||||
if n, err := m.Increment(ctx, "greeting", 1); err == nil || n != 0 || !strings.Contains(err.Error(), "greeting") {
|
||||
t.Fatalf("Increment(non-integer) = %d %v, want an error naming the key", n, err)
|
||||
}
|
||||
if v, _, _ := m.Get(ctx, "greeting"); string(v) != "hello" {
|
||||
t.Fatalf("a failed Increment changed the value: %q", v)
|
||||
}
|
||||
|
||||
// Remember stores the computed value; an error from fn is returned.
|
||||
calls := 0
|
||||
fn := func(context.Context) ([]byte, error) { calls++; return []byte("computed"), nil }
|
||||
for range 3 {
|
||||
if v, err := m.Remember(ctx, "remembered", time.Minute, fn); err != nil || string(v) != "computed" {
|
||||
t.Fatalf("Remember = %q %v", v, err)
|
||||
}
|
||||
}
|
||||
if calls != 1 {
|
||||
t.Fatalf("callback ran %d times, want 1", calls)
|
||||
}
|
||||
boom := errors.New("boom")
|
||||
if v, err := m.Remember(ctx, "failing", time.Minute, func(context.Context) ([]byte, error) { return nil, boom }); v != nil || !errors.Is(err, boom) {
|
||||
t.Fatalf("Remember(failing fn) = %q %v", v, err)
|
||||
}
|
||||
if ok, _ := m.Has(ctx, "failing"); ok {
|
||||
t.Fatal("a failing callback stored a value")
|
||||
}
|
||||
if _, err := m.Remember(ctx, "nilfn", time.Minute, nil); err == nil {
|
||||
t.Fatal("Remember with a nil callback must fail")
|
||||
}
|
||||
|
||||
// Forget and Flush.
|
||||
if err := m.Forget(ctx, "greeting"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ok, _ := m.Has(ctx, "greeting"); ok {
|
||||
t.Fatal("Forget left the key")
|
||||
}
|
||||
if err := m.Flush(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, k := range []string{"hits", "remembered", "copy", "forever0s"} {
|
||||
if ok, _ := m.Has(ctx, k); ok {
|
||||
t.Fatalf("Flush left %q", k)
|
||||
}
|
||||
}
|
||||
|
||||
if _, err := NewMemoryCache("acme:", -1); err == nil || !strings.Contains(err.Error(), CacheMaxEntriesKey) {
|
||||
t.Fatalf("NewMemoryCache(-1) = %v, want an error naming %s", err, CacheMaxEntriesKey)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemoryCacheTTL(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
m := newMemory(t, "acme:", DefaultMaxEntries)
|
||||
|
||||
if err := m.Set(ctx, "short", []byte("x"), 40*time.Millisecond); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ok, _ := m.Has(ctx, "short"); !ok {
|
||||
t.Fatal("a fresh 40 ms entry is missing")
|
||||
}
|
||||
time.Sleep(80 * time.Millisecond)
|
||||
if ok, _ := m.Has(ctx, "short"); ok {
|
||||
t.Fatal("a 40 ms entry outlived 80 ms")
|
||||
}
|
||||
if _, ok, _ := m.Get(ctx, "short"); ok {
|
||||
t.Fatal("Get returned an expired entry")
|
||||
}
|
||||
|
||||
// Increment on an existing key keeps its expiry, as INCRBY does.
|
||||
if err := m.Set(ctx, "counter", []byte("5"), 60*time.Millisecond); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before, _ := m.c.GetIfPresent(m.key("counter"))
|
||||
if n, err := m.Increment(ctx, "counter", 1); err != nil || n != 6 {
|
||||
t.Fatalf("Increment = %d %v", n, err)
|
||||
}
|
||||
after, _ := m.c.GetIfPresent(m.key("counter"))
|
||||
if !after.expiresAt.Equal(before.expiresAt) {
|
||||
t.Fatalf("Increment moved the expiry from %v to %v", before.expiresAt, after.expiresAt)
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
if ok, _ := m.Has(ctx, "counter"); ok {
|
||||
t.Fatal("an incremented entry outlived its TTL")
|
||||
}
|
||||
|
||||
// An expired key restarts at 0 without a TTL.
|
||||
if n, err := m.Increment(ctx, "counter", 2); err != nil || n != 2 {
|
||||
t.Fatalf("Increment after expiry = %d %v, want 2", n, err)
|
||||
}
|
||||
if e, _ := m.c.GetIfPresent(m.key("counter")); !e.expiresAt.IsZero() {
|
||||
t.Fatalf("a restarted counter has expiry %v, want none", e.expiresAt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemoryCacheBounded(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
const limit = 50
|
||||
m := newMemory(t, "acme:", limit)
|
||||
for i := range 5000 {
|
||||
if err := m.Set(ctx, "k"+strconv.Itoa(i), []byte("v"), 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
m.c.CleanUp()
|
||||
if n := m.c.EstimatedSize(); n > limit {
|
||||
t.Fatalf("cache holds %d entries, bound is %d", n, limit)
|
||||
}
|
||||
live := 0
|
||||
for i := range 5000 {
|
||||
if ok, _ := m.Has(ctx, "k"+strconv.Itoa(i)); ok {
|
||||
live++
|
||||
}
|
||||
}
|
||||
if live == 0 || live > limit {
|
||||
t.Fatalf("%d live keys, want between 1 and %d", live, limit)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemoryIncrementConcurrent(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
m := newMemory(t, "acme:", DefaultMaxEntries)
|
||||
if err := m.Set(ctx, "views", []byte("0"), time.Hour); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before, _ := m.c.GetIfPresent(m.key("views"))
|
||||
|
||||
const workers, each = 100, 10
|
||||
var wg sync.WaitGroup
|
||||
start := make(chan struct{})
|
||||
errs := make(chan error, workers*each)
|
||||
for range workers {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
for range each {
|
||||
if _, err := m.Increment(ctx, "views", 1); err != nil {
|
||||
errs <- err
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
close(start)
|
||||
wg.Wait()
|
||||
close(errs)
|
||||
for err := range errs {
|
||||
t.Fatal(err)
|
||||
}
|
||||
v, ok, err := m.Get(ctx, "views")
|
||||
if err != nil || !ok || string(v) != strconv.Itoa(workers*each) {
|
||||
t.Fatalf("views = %q %v %v, want %d", v, ok, err, workers*each)
|
||||
}
|
||||
after, _ := m.c.GetIfPresent(m.key("views"))
|
||||
if !after.expiresAt.Equal(before.expiresAt) {
|
||||
t.Fatalf("concurrent increments moved the expiry from %v to %v", before.expiresAt, after.expiresAt)
|
||||
}
|
||||
}
|
||||
|
||||
// failingCache is a Cache whose reads and writes fail as configured.
|
||||
type failingCache struct {
|
||||
Cache
|
||||
getErr, setErr error
|
||||
sets int
|
||||
}
|
||||
|
||||
func (f *failingCache) Get(context.Context, string) ([]byte, bool, error) {
|
||||
return nil, false, f.getErr
|
||||
}
|
||||
|
||||
func (f *failingCache) Set(context.Context, string, []byte, time.Duration) error {
|
||||
f.sets++
|
||||
return f.setErr
|
||||
}
|
||||
|
||||
func TestRememberDegrades(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
value := func(context.Context) ([]byte, error) { return []byte("fresh"), nil }
|
||||
|
||||
t.Run("a failing read computes, skips the write and warns once", func(t *testing.T) {
|
||||
buf := &syncBuffer{}
|
||||
w := newWarner(slog.New(slog.NewTextHandler(buf, nil)))
|
||||
f := &failingCache{getErr: errors.New("connection refused")}
|
||||
for range 3 {
|
||||
v, err := remember(ctx, f, w, "k", time.Minute, value)
|
||||
if err != nil || string(v) != "fresh" {
|
||||
t.Fatalf("remember = %q %v", v, err)
|
||||
}
|
||||
}
|
||||
if f.sets != 0 {
|
||||
t.Fatalf("a failed read still wrote %d times", f.sets)
|
||||
}
|
||||
lines := buf.lines()
|
||||
if len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], "cache read failed") {
|
||||
t.Fatalf("log = %q, want one read warning", lines)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a failing write returns the value and warns once", func(t *testing.T) {
|
||||
buf := &syncBuffer{}
|
||||
w := newWarner(slog.New(slog.NewTextHandler(buf, nil)))
|
||||
f := &failingCache{setErr: errors.New("read-only replica")}
|
||||
for range 3 {
|
||||
v, err := remember(ctx, f, w, "k", time.Minute, value)
|
||||
if err != nil || string(v) != "fresh" {
|
||||
t.Fatalf("remember = %q %v", v, err)
|
||||
}
|
||||
}
|
||||
if f.sets != 3 {
|
||||
t.Fatalf("%d writes, want 3", f.sets)
|
||||
}
|
||||
lines := buf.lines()
|
||||
if len(lines) != 1 || !strings.Contains(lines[0], "cache write failed") {
|
||||
t.Fatalf("log = %q, want one write warning", lines)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("an error from the callback is returned", func(t *testing.T) {
|
||||
boom := errors.New("boom")
|
||||
f := &failingCache{getErr: errors.New("down")}
|
||||
w := newWarner(slog.New(slog.NewTextHandler(&syncBuffer{}, nil)))
|
||||
if v, err := remember(ctx, f, w, "k", time.Minute, func(context.Context) ([]byte, error) { return nil, boom }); v != nil || !errors.Is(err, boom) {
|
||||
t.Fatalf("remember = %q %v", v, err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a Redis cache whose server is gone still returns the value", func(t *testing.T) {
|
||||
buf := &syncBuffer{}
|
||||
client, err := OpenRedis(ctx, "redis://"+closedAddr(t)+"/0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = client.Close() })
|
||||
c := NewRedisCache(client, "acme:")
|
||||
setLogger(c, slog.New(slog.NewTextHandler(buf, nil)))
|
||||
for range 2 {
|
||||
// The caller's deadline bounds go-redis's dial retries.
|
||||
cctx, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
|
||||
v, err := c.Remember(cctx, "k", time.Minute, value)
|
||||
cancel()
|
||||
if err != nil || string(v) != "fresh" {
|
||||
t.Fatalf("Remember = %q %v", v, err)
|
||||
}
|
||||
}
|
||||
if lines := buf.lines(); len(lines) != 1 || !strings.Contains(lines[0], "cache read failed") {
|
||||
t.Fatalf("log = %q, want one read warning", lines)
|
||||
}
|
||||
})
|
||||
}
|
||||
358
modules/cooler/ratelimit_test.go
Normal file
358
modules/cooler/ratelimit_test.go
Normal file
@@ -0,0 +1,358 @@
|
||||
package cooler_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
"git.golem15.com/golem15/summercms/modules/compass"
|
||||
"git.golem15.com/golem15/summercms/modules/cooler"
|
||||
"git.golem15.com/golem15/summercms/modules/surf"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
)
|
||||
|
||||
// Both stores must satisfy surf.Store; the compiler checks the Redis one here.
|
||||
var _ surf.Store = (*cooler.RateLimitStore)(nil)
|
||||
|
||||
// parityStore is one rate-limit store under test with the way its clock moves.
|
||||
type parityStore struct {
|
||||
name string
|
||||
store surf.Store
|
||||
advance func(time.Duration)
|
||||
}
|
||||
|
||||
// parityStep is one Attempt, after the store's clock moved by advance.
|
||||
type parityStep struct {
|
||||
advance time.Duration
|
||||
key string
|
||||
max int
|
||||
allowed bool
|
||||
attempts int
|
||||
// newWindow: retryAfter is close to the full decay (a window opened).
|
||||
// Otherwise retryAfter never grows within the window and stays at or
|
||||
// below decay minus the time advanced since the window opened.
|
||||
newWindow bool
|
||||
}
|
||||
|
||||
// parityDecay is long enough for real-time stores to cross it in a test and
|
||||
// short enough to keep the table fast.
|
||||
const parityDecay = 300 * time.Millisecond
|
||||
|
||||
// parityTable is the shared fixed-window behaviour of surf.MemoryStore and
|
||||
// the Redis store: the first attempt opens the window, a denial does not
|
||||
// increment, the window never extends, retryAfter never grows and expiry
|
||||
// opens a new window.
|
||||
var parityTable = []parityStep{
|
||||
{key: "login|192.0.2.1", max: 3, allowed: true, attempts: 1, newWindow: true},
|
||||
{key: "login|192.0.2.1", max: 3, allowed: true, attempts: 2},
|
||||
{key: "login|192.0.2.1", max: 3, allowed: true, attempts: 3},
|
||||
{key: "login|192.0.2.1", max: 3, allowed: false, attempts: 3},
|
||||
{key: "login|192.0.2.1", max: 3, allowed: false, attempts: 3},
|
||||
// Another key has its own window.
|
||||
{key: "login|192.0.2.2", max: 3, allowed: true, attempts: 1, newWindow: true},
|
||||
// A max of 0 denies without counting.
|
||||
{key: "closed", max: 0, allowed: false, attempts: 0, newWindow: true},
|
||||
// Inside the window: still denied, the window did not move.
|
||||
{advance: 120 * time.Millisecond, key: "login|192.0.2.1", max: 3, allowed: false, attempts: 3},
|
||||
// Past the window: a new one opens at 1.
|
||||
{advance: 250 * time.Millisecond, key: "login|192.0.2.1", max: 3, allowed: true, attempts: 1, newWindow: true},
|
||||
{key: "login|192.0.2.1", max: 3, allowed: true, attempts: 2},
|
||||
}
|
||||
|
||||
// runParity drives the table through s and checks every answer.
|
||||
func runParity(t *testing.T, s parityStore) {
|
||||
t.Helper()
|
||||
opened := map[string]time.Duration{} // advanced since the key's window opened
|
||||
last := map[string]time.Duration{} // last retryAfter per key
|
||||
for i, step := range parityTable {
|
||||
if step.advance > 0 {
|
||||
s.advance(step.advance)
|
||||
for k := range opened {
|
||||
opened[k] += step.advance
|
||||
}
|
||||
}
|
||||
allowed, attempts, retry := s.store.Attempt(step.key, step.max, parityDecay)
|
||||
if allowed != step.allowed || attempts != step.attempts {
|
||||
t.Fatalf("%s step %d (%s max %d): allowed=%v attempts=%d, want %v %d",
|
||||
s.name, i, step.key, step.max, allowed, attempts, step.allowed, step.attempts)
|
||||
}
|
||||
if retry <= 0 || retry > parityDecay {
|
||||
t.Fatalf("%s step %d: retryAfter %s outside (0, %s]", s.name, i, retry, parityDecay)
|
||||
}
|
||||
if step.newWindow {
|
||||
if retry < parityDecay-100*time.Millisecond {
|
||||
t.Fatalf("%s step %d: a new window answered retryAfter %s, want about %s", s.name, i, retry, parityDecay)
|
||||
}
|
||||
opened[step.key] = 0
|
||||
} else {
|
||||
if prev, ok := last[step.key]; ok && retry > prev {
|
||||
t.Fatalf("%s step %d: retryAfter grew from %s to %s", s.name, i, prev, retry)
|
||||
}
|
||||
if limit := parityDecay - opened[step.key]; retry > limit {
|
||||
t.Fatalf("%s step %d: retryAfter %s, want at most %s (the window extended)", s.name, i, retry, limit)
|
||||
}
|
||||
}
|
||||
last[step.key] = retry
|
||||
}
|
||||
}
|
||||
|
||||
func sleepFor(d time.Duration) { time.Sleep(d) }
|
||||
|
||||
// miniStore returns a Redis rate-limit store on a fresh miniredis.
|
||||
func miniStore(t *testing.T, prefix string, fallback cooler.Attempter) (*miniredis.Miniredis, *cooler.RateLimitStore) {
|
||||
t.Helper()
|
||||
mr := miniredis.RunT(t)
|
||||
client, err := cooler.OpenRedis(t.Context(), "redis://"+mr.Addr()+"/0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = client.Close() })
|
||||
return mr, cooler.NewRateLimitStore(client, prefix, fallback)
|
||||
}
|
||||
|
||||
func TestRateLimitParity(t *testing.T) {
|
||||
t.Run("surf.MemoryStore", func(t *testing.T) {
|
||||
mem := surf.NewMemoryStore(0)
|
||||
runParity(t, parityStore{name: "memory", store: mem, advance: sleepFor})
|
||||
})
|
||||
t.Run("Redis store on miniredis", func(t *testing.T) {
|
||||
mr, store := miniStore(t, "acme:", nil)
|
||||
runParity(t, parityStore{name: "miniredis", store: store, advance: mr.FastForward})
|
||||
})
|
||||
}
|
||||
|
||||
// countAdmitted runs workers concurrent attempts on one key and returns how
|
||||
// many were admitted.
|
||||
func countAdmitted(s surf.Store, workers, max int) int {
|
||||
var admitted atomic.Int64
|
||||
var wg sync.WaitGroup
|
||||
start := make(chan struct{})
|
||||
for range workers {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
if ok, _, _ := s.Attempt("burst", max, time.Minute); ok {
|
||||
admitted.Add(1)
|
||||
}
|
||||
}()
|
||||
}
|
||||
close(start)
|
||||
wg.Wait()
|
||||
return int(admitted.Load())
|
||||
}
|
||||
|
||||
func TestRateLimitConcurrentNeverExceedsMax(t *testing.T) {
|
||||
const workers, max = 50, 10
|
||||
if got := countAdmitted(surf.NewMemoryStore(0), workers, max); got != max {
|
||||
t.Fatalf("MemoryStore admitted %d of %d, want %d", got, workers, max)
|
||||
}
|
||||
mr, store := miniStore(t, "acme:", nil)
|
||||
if got := countAdmitted(store, workers, max); got != max {
|
||||
t.Fatalf("Redis store admitted %d of %d, want %d", got, workers, max)
|
||||
}
|
||||
if got, err := mr.Get("acme:rl:burst"); err != nil || got != fmt.Sprint(max) {
|
||||
t.Fatalf("Redis counter = %q %v, want %d (denials must not increment)", got, err, max)
|
||||
}
|
||||
}
|
||||
|
||||
// captureDefault routes slog.Default to a buffer for the test. Stores built
|
||||
// after the call log there.
|
||||
func captureDefault(t *testing.T) *lockedBuffer {
|
||||
t.Helper()
|
||||
buf := &lockedBuffer{}
|
||||
prev := slog.Default()
|
||||
slog.SetDefault(slog.New(slog.NewTextHandler(buf, nil)))
|
||||
t.Cleanup(func() { slog.SetDefault(prev) })
|
||||
return buf
|
||||
}
|
||||
|
||||
type lockedBuffer struct {
|
||||
mu sync.Mutex
|
||||
buf bytes.Buffer
|
||||
}
|
||||
|
||||
func (b *lockedBuffer) Write(p []byte) (int, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.Write(p)
|
||||
}
|
||||
|
||||
func (b *lockedBuffer) lines() []string {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
var out []string
|
||||
for _, l := range strings.Split(strings.TrimSpace(b.buf.String()), "\n") {
|
||||
if l != "" {
|
||||
out = append(out, l)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestRateLimitFallbackOnError(t *testing.T) {
|
||||
t.Run("a stopped Redis counts in the fallback and warns once", func(t *testing.T) {
|
||||
logs := captureDefault(t)
|
||||
fallback := surf.NewMemoryStore(0)
|
||||
mr, store := miniStore(t, "acme:", fallback)
|
||||
if ok, n, _ := store.Attempt("login", 2, time.Minute); !ok || n != 1 {
|
||||
t.Fatalf("with Redis up: %v %d", ok, n)
|
||||
}
|
||||
addr := mr.Addr()
|
||||
mr.Close()
|
||||
|
||||
want := []struct {
|
||||
allowed bool
|
||||
attempts int
|
||||
}{{true, 1}, {true, 2}, {false, 2}}
|
||||
for i, w := range want {
|
||||
start := time.Now()
|
||||
ok, n, retry := store.Attempt("login", 2, time.Minute)
|
||||
if elapsed := time.Since(start); elapsed > cooler.AttemptTimeout+300*time.Millisecond {
|
||||
t.Fatalf("attempt %d took %s, want at most about %s", i, elapsed, cooler.AttemptTimeout)
|
||||
}
|
||||
if ok != w.allowed || n != w.attempts || retry <= 0 {
|
||||
t.Fatalf("fallback attempt %d = %v %d %s, want %v %d", i, ok, n, retry, w.allowed, w.attempts)
|
||||
}
|
||||
}
|
||||
// The counts live in the fallback store itself.
|
||||
if ok, n, _ := fallback.Attempt("login", 2, time.Minute); ok || n != 2 {
|
||||
t.Fatalf("fallback store = %v %d, want denied at 2", ok, n)
|
||||
}
|
||||
lines := logs.lines()
|
||||
if len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") ||
|
||||
!strings.Contains(lines[0], "rate-limit store failed") || !strings.Contains(lines[0], "endpoint="+addr) {
|
||||
t.Fatalf("log = %q, want one warning naming the endpoint", lines)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("without a fallback an outage admits", func(t *testing.T) {
|
||||
captureDefault(t)
|
||||
mr, store := miniStore(t, "acme:", nil)
|
||||
mr.Close()
|
||||
for range 3 {
|
||||
if ok, n, retry := store.Attempt("login", 1, time.Minute); !ok || n != 0 || retry != 0 {
|
||||
t.Fatalf("Attempt = %v %d %s, want true 0 0", ok, n, retry)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a store without a client uses the fallback", func(t *testing.T) {
|
||||
captureDefault(t)
|
||||
store := cooler.NewRateLimitStore(nil, "acme:", surf.NewMemoryStore(0))
|
||||
if ok, n, _ := store.Attempt("k", 1, time.Minute); !ok || n != 1 {
|
||||
t.Fatalf("first = %v %d", ok, n)
|
||||
}
|
||||
if ok, n, _ := store.Attempt("k", 1, time.Minute); ok || n != 1 {
|
||||
t.Fatalf("second = %v %d", ok, n)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a script error answers from the fallback", func(t *testing.T) {
|
||||
captureDefault(t)
|
||||
mr, store := miniStore(t, "acme:", surf.NewMemoryStore(0))
|
||||
mr.SetError("LOADING Redis is loading the dataset in memory")
|
||||
if ok, n, _ := store.Attempt("k", 1, time.Minute); !ok || n != 1 {
|
||||
t.Fatalf("Attempt = %v %d", ok, n)
|
||||
}
|
||||
mr.SetError("")
|
||||
if ok, n, _ := store.Attempt("k", 1, time.Minute); !ok || n != 1 {
|
||||
t.Fatalf("after recovery Redis counts again from 1: %v %d", ok, n)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestRateLimitKeyPrefix(t *testing.T) {
|
||||
mr, store := miniStore(t, "acme:", nil)
|
||||
store.Attempt("login|192.0.2.1", 5, 90*time.Second)
|
||||
if got, err := mr.Get("acme:rl:login|192.0.2.1"); err != nil || got != "1" {
|
||||
t.Fatalf("acme:rl:login|192.0.2.1 = %q %v", got, err)
|
||||
}
|
||||
if ttl := mr.TTL("acme:rl:login|192.0.2.1"); ttl != 90*time.Second {
|
||||
t.Fatalf("window TTL = %s, want 90s", ttl)
|
||||
}
|
||||
// A sub-millisecond decay is clamped to 1 ms rather than failing.
|
||||
if ok, _, _ := store.Attempt("tiny", 1, time.Microsecond); !ok {
|
||||
t.Fatal("a sub-ms decay must still admit")
|
||||
}
|
||||
if ttl := mr.TTL("acme:rl:tiny"); ttl != time.Millisecond {
|
||||
t.Fatalf("sub-ms window TTL = %s, want 1ms", ttl)
|
||||
}
|
||||
if keys := mr.Keys(); len(keys) != 2 {
|
||||
t.Fatalf("keys = %q, want only the two rl keys", keys)
|
||||
}
|
||||
|
||||
// Through Install: the prefix derives from app.name and the store from
|
||||
// http.rate_limit.store.
|
||||
cfg := emptyConfig(t)
|
||||
mustSet(t, cfg, "app.name", "Acme Shop")
|
||||
mustSet(t, cfg, cooler.RateLimitStoreKey, "redis")
|
||||
mustSet(t, cfg, cooler.RedisURLKey, "redis://"+mr.Addr()+"/0")
|
||||
app := backpack.New(cfg)
|
||||
h, err := cooler.Install(context.Background(), app)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = h.Close() })
|
||||
if !h.RateLimitRedis() {
|
||||
t.Fatal("RateLimitRedis = false with http.rate_limit.store redis")
|
||||
}
|
||||
fallback := surf.NewMemoryStore(0)
|
||||
s := h.RateLimitStore(fallback)
|
||||
if s == cooler.Attempter(fallback) {
|
||||
t.Fatal("RateLimitStore returned the fallback with the Redis store selected")
|
||||
}
|
||||
s.Attempt("login", 1, time.Minute)
|
||||
if !mr.Exists("acme_shop:rl:login") {
|
||||
t.Fatalf("keys = %q, want acme_shop:rl:login", mr.Keys())
|
||||
}
|
||||
|
||||
// cache.prefix wins over app.name.
|
||||
mustSet(t, cfg, cooler.CachePrefixKey, "shop1:")
|
||||
h2, err := cooler.Install(context.Background(), backpack.New(cfg))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = h2.Close() })
|
||||
h2.RateLimitStore(fallback).Attempt("login", 1, time.Minute)
|
||||
if !mr.Exists("shop1:rl:login") {
|
||||
t.Fatalf("keys = %q, want shop1:rl:login", mr.Keys())
|
||||
}
|
||||
}
|
||||
|
||||
// emptyConfig returns a config that ignores the process environment.
|
||||
func emptyConfig(t *testing.T) *compass.Config {
|
||||
t.Helper()
|
||||
cfg, err := compass.Open(compass.Options{Dir: t.TempDir(), Environ: []string{}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
func mustSet(t *testing.T, cfg *compass.Config, key string, value any) {
|
||||
t.Helper()
|
||||
if err := cfg.Set(key, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// deadAddr returns a localhost address nothing listens on.
|
||||
func deadAddr(t *testing.T) string {
|
||||
t.Helper()
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
addr := ln.Addr().String()
|
||||
_ = ln.Close()
|
||||
return addr
|
||||
}
|
||||
200
modules/cooler/redis_test.go
Normal file
200
modules/cooler/redis_test.go
Normal file
@@ -0,0 +1,200 @@
|
||||
package cooler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// closedAddr returns a localhost address nothing listens on.
|
||||
func closedAddr(t *testing.T) string {
|
||||
t.Helper()
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
addr := ln.Addr().String()
|
||||
_ = ln.Close()
|
||||
return addr
|
||||
}
|
||||
|
||||
// miniClient starts a miniredis and returns it with a client from OpenRedis.
|
||||
func miniClient(t *testing.T) (*miniredis.Miniredis, *redis.Client) {
|
||||
t.Helper()
|
||||
mr := miniredis.RunT(t)
|
||||
client, err := OpenRedis(t.Context(), "redis://"+mr.Addr()+"/0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = client.Close() })
|
||||
return mr, client
|
||||
}
|
||||
|
||||
func TestRedisCache(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
mr, client := miniClient(t)
|
||||
c := NewRedisCache(client, "acme:")
|
||||
|
||||
if v, ok, err := c.Get(ctx, "missing"); v != nil || ok || err != nil {
|
||||
t.Fatalf("Get(missing) = %q %v %v", v, ok, err)
|
||||
}
|
||||
if ok, err := c.Has(ctx, "missing"); ok || err != nil {
|
||||
t.Fatalf("Has(missing) = %v %v", ok, err)
|
||||
}
|
||||
|
||||
// Keys live under <prefix>cache:<key>; a TTL is stored with PX.
|
||||
if err := c.Set(ctx, "greeting", []byte("hello"), 10*time.Second); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, err := mr.Get("acme:cache:greeting"); err != nil || got != "hello" {
|
||||
t.Fatalf("raw acme:cache:greeting = %q %v", got, err)
|
||||
}
|
||||
if ttl := mr.TTL("acme:cache:greeting"); ttl != 10*time.Second {
|
||||
t.Fatalf("TTL = %s, want 10s", ttl)
|
||||
}
|
||||
if v, ok, err := c.Get(ctx, "greeting"); err != nil || !ok || string(v) != "hello" {
|
||||
t.Fatalf("Get = %q %v %v", v, ok, err)
|
||||
}
|
||||
mr.FastForward(9 * time.Second)
|
||||
if ok, _ := c.Has(ctx, "greeting"); !ok {
|
||||
t.Fatal("the key expired before its TTL")
|
||||
}
|
||||
mr.FastForward(2 * time.Second)
|
||||
if ok, _ := c.Has(ctx, "greeting"); ok {
|
||||
t.Fatal("the key outlived its TTL")
|
||||
}
|
||||
|
||||
// A sub-millisecond TTL is rounded up to 1 ms, not stored forever.
|
||||
if err := c.Set(ctx, "tiny", []byte("x"), time.Microsecond); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ttl := mr.TTL("acme:cache:tiny"); ttl != time.Millisecond {
|
||||
t.Fatalf("sub-ms TTL stored as %s, want 1ms", ttl)
|
||||
}
|
||||
|
||||
// ttl 0 and a negative ttl store without expiry; a nil value stores empty.
|
||||
for _, ttl := range []time.Duration{0, -time.Second} {
|
||||
if err := c.Set(ctx, "forever", nil, ttl); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := mr.TTL("acme:cache:forever"); got != 0 {
|
||||
t.Fatalf("ttl %s stored TTL %s, want none", ttl, got)
|
||||
}
|
||||
if v, ok, err := c.Get(ctx, "forever"); err != nil || !ok || len(v) != 0 {
|
||||
t.Fatalf("Get(forever) = %q %v %v", v, ok, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Increment: INCRBY from missing has no TTL, an existing key keeps its TTL.
|
||||
if n, err := c.Increment(ctx, "hits", 4); err != nil || n != 4 {
|
||||
t.Fatalf("Increment(missing) = %d %v", n, err)
|
||||
}
|
||||
if ttl := mr.TTL("acme:cache:hits"); ttl != 0 {
|
||||
t.Fatalf("Increment(missing) set TTL %s", ttl)
|
||||
}
|
||||
if err := c.Set(ctx, "views", []byte("10"), time.Minute); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n, err := c.Increment(ctx, "views", 5); err != nil || n != 15 {
|
||||
t.Fatalf("Increment(existing) = %d %v", n, err)
|
||||
}
|
||||
if ttl := mr.TTL("acme:cache:views"); ttl != time.Minute {
|
||||
t.Fatalf("Increment changed the TTL to %s", ttl)
|
||||
}
|
||||
_ = c.Set(ctx, "word", []byte("abc"), 0)
|
||||
if _, err := c.Increment(ctx, "word", 1); err == nil {
|
||||
t.Fatal("Increment on a non-integer must fail")
|
||||
}
|
||||
|
||||
// Remember stores with the TTL and calls the callback once.
|
||||
calls := 0
|
||||
fn := func(context.Context) ([]byte, error) { calls++; return []byte("computed"), nil }
|
||||
for range 3 {
|
||||
if v, err := c.Remember(ctx, "remembered", 30*time.Second, fn); err != nil || string(v) != "computed" {
|
||||
t.Fatalf("Remember = %q %v", v, err)
|
||||
}
|
||||
}
|
||||
if calls != 1 || mr.TTL("acme:cache:remembered") != 30*time.Second {
|
||||
t.Fatalf("Remember: %d calls, TTL %s", calls, mr.TTL("acme:cache:remembered"))
|
||||
}
|
||||
|
||||
// Forget removes only that key.
|
||||
if err := c.Forget(ctx, "views"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if mr.Exists("acme:cache:views") || !mr.Exists("acme:cache:hits") {
|
||||
t.Fatal("Forget removed the wrong keys")
|
||||
}
|
||||
|
||||
// A cache without a client answers errNilClient on every method.
|
||||
var none Cache = &redisCache{}
|
||||
if _, _, err := none.Get(ctx, "k"); !errors.Is(err, errNilClient) {
|
||||
t.Fatalf("Get without client = %v", err)
|
||||
}
|
||||
if err := none.Set(ctx, "k", nil, 0); !errors.Is(err, errNilClient) {
|
||||
t.Fatalf("Set without client = %v", err)
|
||||
}
|
||||
if _, err := none.Has(ctx, "k"); !errors.Is(err, errNilClient) {
|
||||
t.Fatalf("Has without client = %v", err)
|
||||
}
|
||||
if err := none.Forget(ctx, "k"); !errors.Is(err, errNilClient) {
|
||||
t.Fatalf("Forget without client = %v", err)
|
||||
}
|
||||
if _, err := none.Increment(ctx, "k", 1); !errors.Is(err, errNilClient) {
|
||||
t.Fatalf("Increment without client = %v", err)
|
||||
}
|
||||
if err := none.Flush(ctx); !errors.Is(err, errNilClient) {
|
||||
t.Fatalf("Flush without client = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRedisFlushOnlyPrefix(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
mr, client := miniClient(t)
|
||||
|
||||
// A prefix with glob characters is matched literally.
|
||||
const prefix = "a*[b]?:"
|
||||
c := NewRedisCache(client, prefix)
|
||||
for i := range 20 {
|
||||
if err := c.Set(ctx, "k"+strconv.Itoa(i), []byte("v"), 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
survivors := map[string]string{
|
||||
prefix + "rl:login|192.0.2.1": "3", // this app's rate-limit counter
|
||||
"other:cache:greeting": "theirs", // another app's cache
|
||||
"aXbY:cache:k1": "glob", // matches the prefix as a pattern
|
||||
"a*[b]?:cachex": "near", // shares the prefix, not the cache: part
|
||||
"plain": "x",
|
||||
}
|
||||
for k, v := range survivors {
|
||||
if err := mr.Set(k, v); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := c.Flush(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := range 20 {
|
||||
if mr.Exists(prefix + "cache:k" + strconv.Itoa(i)) {
|
||||
t.Fatalf("Flush left %scache:k%d", prefix, i)
|
||||
}
|
||||
}
|
||||
for k, v := range survivors {
|
||||
if got, err := mr.Get(k); err != nil || got != v {
|
||||
t.Fatalf("Flush removed or changed %q: %q %v", k, got, err)
|
||||
}
|
||||
}
|
||||
if err := c.Flush(ctx); err != nil {
|
||||
t.Fatalf("Flush on an empty cache: %v", err)
|
||||
}
|
||||
if got := escapeGlob(`a*b?[c]\`); got != `a\*b\?\[c\]\\` {
|
||||
t.Fatalf("escapeGlob = %q", got)
|
||||
}
|
||||
}
|
||||
@@ -3,7 +3,9 @@ package cooler
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
@@ -16,6 +18,11 @@ import (
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
"git.golem15.com/golem15/summercms/modules/compass"
|
||||
"git.golem15.com/golem15/summercms/modules/lagoon"
|
||||
_ "github.com/jackc/pgx/v5/stdlib"
|
||||
"github.com/testcontainers/testcontainers-go"
|
||||
"github.com/testcontainers/testcontainers-go/modules/postgres"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -383,3 +390,252 @@ func testConfig(t *testing.T) *compass.Config {
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// purgeDB starts a Postgres container for the test and returns a GORM handle
|
||||
// with a table to write in transactions. It skips under -short.
|
||||
func purgeDB(t *testing.T) *gorm.DB {
|
||||
t.Helper()
|
||||
if testing.Short() {
|
||||
t.Skip("requires testcontainers postgres")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
t.Cleanup(cancel)
|
||||
ctr, err := postgres.Run(ctx,
|
||||
"postgres:16-alpine",
|
||||
postgres.WithDatabase("cooler"),
|
||||
postgres.WithUsername("cooler"),
|
||||
postgres.WithPassword("cooler"),
|
||||
postgres.BasicWaitStrategies(),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("postgres: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = testcontainers.TerminateContainer(ctr) })
|
||||
dsn, err := ctr.ConnectionString(ctx, "sslmode=disable")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db, err := sql.Open("pgx", dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
if err := db.PingContext(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
gdb, err := lagoon.Use(ctx, db)
|
||||
if err != nil {
|
||||
t.Fatalf("lagoon.Use: %v", err)
|
||||
}
|
||||
if err := gdb.Exec(`CREATE TABLE purge_posts (id serial PRIMARY KEY, slug text NOT NULL)`).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return gdb
|
||||
}
|
||||
|
||||
func countPosts(t *testing.T, gdb *gorm.DB, slug string) int64 {
|
||||
t.Helper()
|
||||
var n int64
|
||||
if err := gdb.Raw(`SELECT count(*) FROM purge_posts WHERE slug = ?`, slug).Scan(&n).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func TestSiteCacheAfterCommitOnly(t *testing.T) {
|
||||
gdb := purgeDB(t)
|
||||
captureLogs(t)
|
||||
srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(`{"purged":1}`)) })
|
||||
p := newTestPurger(t, srv)
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("a rolled-back save sends nothing", func(t *testing.T) {
|
||||
boom := errors.New("validation failed")
|
||||
err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
if err := tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('rolled-back')`).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "rolled-back")
|
||||
return boom
|
||||
})
|
||||
if !errors.Is(err, boom) {
|
||||
t.Fatalf("Transaction = %v, want the callback's error", err)
|
||||
}
|
||||
if n := countPosts(t, gdb, "rolled-back"); n != 0 {
|
||||
t.Fatalf("%d rows after rollback", n)
|
||||
}
|
||||
if got := hits(); len(got) != 0 {
|
||||
t.Fatalf("a rolled-back save sent %d purges", len(got))
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a committed save sends one purge after the commit", func(t *testing.T) {
|
||||
var during int
|
||||
err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
if err := tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('committed')`).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "committed")
|
||||
during = len(hits())
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if during != 0 {
|
||||
t.Fatalf("%d purges were sent before the commit", during)
|
||||
}
|
||||
got := hits()
|
||||
if len(got) != 1 || got[0].body["kind"] != "post" || got[0].body["slug"] != "committed" {
|
||||
t.Fatalf("purges = %#v, want one for post committed", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a nested transaction purges once, after the outer commit", func(t *testing.T) {
|
||||
before := len(hits())
|
||||
var afterInner int
|
||||
err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
if err := lagoon.Transaction(ctx, tx, func(ctx context.Context, inner *gorm.DB) error {
|
||||
PurgeSiteCacheAfterCommit(ctx, inner, p, "category", "nested")
|
||||
return inner.Exec(`INSERT INTO purge_posts (slug) VALUES ('nested')`).Error
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
afterInner = len(hits())
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if afterInner != before {
|
||||
t.Fatal("the purge ran when the inner transaction finished")
|
||||
}
|
||||
if got := hits(); len(got) != before+1 || got[len(got)-1].body["slug"] != "nested" {
|
||||
t.Fatalf("purges = %#v, want one more for nested", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestSiteCacheFailureNeverFails(t *testing.T) {
|
||||
gdb := purgeDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("a 500 answer leaves the save committed", func(t *testing.T) {
|
||||
logs := captureLogs(t)
|
||||
srv, hits := purgeReceiver(t, func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) })
|
||||
p := newTestPurger(t, srv)
|
||||
err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "server-error")
|
||||
return tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('server-error')`).Error
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("a failing purge failed the save: %v", err)
|
||||
}
|
||||
if n := countPosts(t, gdb, "server-error"); n != 1 || len(hits()) != 1 {
|
||||
t.Fatalf("rows %d, purges %d; want 1 and 1", n, len(hits()))
|
||||
}
|
||||
if lines := logs.lines(); len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], "status=500") {
|
||||
t.Fatalf("log = %q", lines)
|
||||
}
|
||||
assertNoToken(t, logs)
|
||||
})
|
||||
|
||||
t.Run("a slow receiver is abandoned and the save stays committed", func(t *testing.T) {
|
||||
logs := captureLogs(t)
|
||||
release := make(chan struct{})
|
||||
srv, _ := purgeReceiver(t, func(_ http.ResponseWriter, r *http.Request) {
|
||||
select {
|
||||
case <-release:
|
||||
case <-r.Context().Done():
|
||||
}
|
||||
})
|
||||
defer close(release)
|
||||
p := newTestPurger(t, srv)
|
||||
start := time.Now()
|
||||
err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "slow")
|
||||
return tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('slow')`).Error
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("a slow purge failed the save: %v", err)
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed > SiteCachePurgeTimeout+2*time.Second {
|
||||
t.Fatalf("the save waited %s, want the purge bounded by %s", elapsed, SiteCachePurgeTimeout)
|
||||
}
|
||||
if n := countPosts(t, gdb, "slow"); n != 1 {
|
||||
t.Fatalf("%d rows, want the save committed", n)
|
||||
}
|
||||
if lines := logs.lines(); len(lines) != 1 || !strings.Contains(lines[0], "error=timeout") {
|
||||
t.Fatalf("log = %q", lines)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("an unreachable receiver leaves the save committed", func(t *testing.T) {
|
||||
captureLogs(t)
|
||||
srv, _ := purgeReceiver(t, func(http.ResponseWriter, *http.Request) {})
|
||||
p := newTestPurger(t, srv)
|
||||
srv.Close()
|
||||
err := lagoon.Transaction(ctx, gdb, func(ctx context.Context, tx *gorm.DB) error {
|
||||
PurgeSiteCacheAfterCommit(ctx, tx, p, "post", "unreachable")
|
||||
return tx.Exec(`INSERT INTO purge_posts (slug) VALUES ('unreachable')`).Error
|
||||
})
|
||||
if err != nil || countPosts(t, gdb, "unreachable") != 1 {
|
||||
t.Fatalf("Transaction = %v, rows %d", err, countPosts(t, gdb, "unreachable"))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestSiteCacheFromApp(t *testing.T) {
|
||||
if p := SiteCacheFromApp(nil); p != nil {
|
||||
t.Fatal("SiteCacheFromApp(nil) must be nil")
|
||||
}
|
||||
if p := SiteCacheFromApp(backpack.New(nil)); p != nil {
|
||||
t.Fatal("an app without config has no purger")
|
||||
}
|
||||
|
||||
cfg := testConfig(t)
|
||||
app := backpack.New(cfg)
|
||||
if p := SiteCacheFromApp(app); p != nil {
|
||||
t.Fatalf("unset keys: %v, want nil", p)
|
||||
}
|
||||
if _, ok := app.Lookup[*SiteCachePurger](); ok {
|
||||
t.Fatal("unset keys published a purger")
|
||||
}
|
||||
if err := cfg.Set(SiteCachePurgeURLKey, "https://www.example.test/api/__site__/cache/purge"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := cfg.Set(SiteCachePurgeTokenKey, purgeToken40); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Concurrent first calls build and publish one purger.
|
||||
var wg sync.WaitGroup
|
||||
got := make([]*SiteCachePurger, 20)
|
||||
for i := range got {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
got[i] = SiteCacheFromApp(app)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
for i, p := range got {
|
||||
if p == nil || p != got[0] {
|
||||
t.Fatalf("call %d = %v, want the one published purger", i, p)
|
||||
}
|
||||
}
|
||||
if pub, ok := app.Lookup[*SiteCachePurger](); !ok || pub != got[0] {
|
||||
t.Fatal("the purger was not published")
|
||||
}
|
||||
if got[0].Endpoint() != "www.example.test/api/__site__/cache/purge" {
|
||||
t.Fatalf("Endpoint = %q", got[0].Endpoint())
|
||||
}
|
||||
|
||||
// A later config change does not rebuild the published purger.
|
||||
if err := cfg.Set(SiteCachePurgeURLKey, "https://other.example.test/purge"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if again := SiteCacheFromApp(app); again != got[0] {
|
||||
t.Fatal("SiteCacheFromApp rebuilt a published purger")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -91,11 +91,32 @@ ATTACH_TESTS=(
|
||||
TestPublicURLValidation
|
||||
)
|
||||
|
||||
# cooler: caches, rate-limit parity with surf.MemoryStore (miniredis),
|
||||
# Install, FromApp and the site-cache purger.
|
||||
COOLER_TESTS=(
|
||||
TestMemoryCache TestMemoryCacheTTL TestMemoryCacheBounded
|
||||
TestMemoryIncrementConcurrent TestRememberDegrades
|
||||
TestRedisCache TestRedisFlushOnlyPrefix
|
||||
TestRateLimitParity
|
||||
TestRateLimitConcurrentNeverExceedsMax
|
||||
TestRateLimitFallbackOnError TestRateLimitKeyPrefix
|
||||
TestInstallConfigErrors TestInstallPingWarning TestInstallReusesFromApp
|
||||
TestFromAppLazy TestInstallSiteCacheValidation
|
||||
TestSiteCachePurgeConfig TestSiteCachePurgeRequest TestSiteCacheFromApp
|
||||
)
|
||||
|
||||
# docker stage: testcontainers suites.
|
||||
DOCKER_CABANA_TESTS=(TestReorderHookRunsInTx TestReorderHookErrorRollsBack TestReorderWithoutHookUnchanged)
|
||||
DOCKER_ATTACH_TESTS=(
|
||||
TestStoreDiskExtension
|
||||
)
|
||||
# Postgres for the after-commit purge, redis:7-alpine for the rest.
|
||||
DOCKER_COOLER_TESTS=(
|
||||
TestRateLimitParityDocker
|
||||
TestRedisCacheDocker
|
||||
TestSiteCacheAfterCommitOnly
|
||||
TestSiteCacheFailureNeverFails
|
||||
)
|
||||
|
||||
usage() {
|
||||
cat >&2 <<'EOF'
|
||||
@@ -363,6 +384,7 @@ run_go() {
|
||||
go_named "$ROOT" ./internal/build -short "${BUILD_TESTS[@]}"
|
||||
go_named "$ROOT" ./modules/lagoon/attach -short "${ATTACH_TESTS[@]}"
|
||||
go_named "$ROOT" ./modules/lagoon/attach "-short -tags summercms_s3" "${ATTACH_TESTS[@]}"
|
||||
go_named "$ROOT" ./modules/cooler -short "${COOLER_TESTS[@]}"
|
||||
echo "phase19.3 go passed"
|
||||
}
|
||||
|
||||
@@ -377,6 +399,7 @@ run_docker() {
|
||||
fi
|
||||
go_named "$ROOT" ./modules/cabana "-timeout 20m" "${DOCKER_CABANA_TESTS[@]}"
|
||||
go_named "$ROOT" ./modules/lagoon/attach "-timeout 20m" "${DOCKER_ATTACH_TESTS[@]}"
|
||||
go_named "$ROOT" ./modules/cooler "-timeout 20m" "${DOCKER_COOLER_TESTS[@]}"
|
||||
echo "phase19.3 docker passed"
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user