diff --git a/go.mod b/go.mod index 104f3dd..0c95a73 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 5276e93..e10725b 100644 --- a/go.sum +++ b/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= diff --git a/modules/cooler/README.md b/modules/cooler/README.md index 9c8dc5a..b4f2413 100644 --- a/modules/cooler/README.md +++ b/modules/cooler/README.md @@ -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`). diff --git a/modules/cooler/docker_test.go b/modules/cooler/docker_test.go new file mode 100644 index 0000000..1b808d2 --- /dev/null +++ b/modules/cooler/docker_test.go @@ -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 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) + } + } +} diff --git a/modules/cooler/install_test.go b/modules/cooler/install_test.go new file mode 100644 index 0000000..7c9b79d --- /dev/null +++ b/modules/cooler/install_test.go @@ -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) + } + }) +} diff --git a/modules/cooler/memory_test.go b/modules/cooler/memory_test.go new file mode 100644 index 0000000..822a3d9 --- /dev/null +++ b/modules/cooler/memory_test.go @@ -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 cache:. + 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) + } + }) +} diff --git a/modules/cooler/ratelimit_test.go b/modules/cooler/ratelimit_test.go new file mode 100644 index 0000000..cebc46e --- /dev/null +++ b/modules/cooler/ratelimit_test.go @@ -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 +} diff --git a/modules/cooler/redis_test.go b/modules/cooler/redis_test.go new file mode 100644 index 0000000..6120cfa --- /dev/null +++ b/modules/cooler/redis_test.go @@ -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 cache:; 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) + } +} diff --git a/modules/cooler/sitecache_test.go b/modules/cooler/sitecache_test.go index b66a9e2..b1de2b3 100644 --- a/modules/cooler/sitecache_test.go +++ b/modules/cooler/sitecache_test.go @@ -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") + } +} diff --git a/scripts/check-phase19.3.sh b/scripts/check-phase19.3.sh index d6da811..0f44da1 100755 --- a/scripts/check-phase19.3.sh +++ b/scripts/check-phase19.3.sh @@ -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" }