diff --git a/.github/workflows/source.yml b/.github/workflows/source.yml index ff89200..de6758e 100644 --- a/.github/workflows/source.yml +++ b/.github/workflows/source.yml @@ -61,6 +61,7 @@ jobs: module: - source/kafka - source/jetstream + - source/redis - source/cloudevents - examples/sourcedrive runs-on: ${{ matrix.os }} @@ -108,6 +109,7 @@ jobs: - source/statemachine - source/kafka - source/jetstream + - source/redis - source/cloudevents - examples/sourcedrive runs-on: ubuntu-latest diff --git a/README.md b/README.md index 24ce935..c4f66f3 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,7 @@ stability label. | `source` | Ingress seam: consume streams and drive statecharts; ack on durable transition. | experimental | | `source/kafka` | Kafka/RedPanda Inlet over franz-go: group consumer, mark-commit-after-process. | experimental | | `source/jetstream` | NATS JetStream Inlet over nats.go: pull consumer, ack/nak/term, MaxAckPending. | experimental | +| `source/redis` | Redis Streams Inlet over go-redis: consumer group, XACK/pending-claim, DLQ. | experimental | | `source/cloudevents` | CloudEvents codec with structured and binary content modes. | experimental | | `source/statemachine` | Bridge: an inbound message drives a transition, ack tied to the durable commit. | experimental | @@ -81,9 +82,9 @@ snapshots; inspection; and JSON (de)serialization. It is backed by its `analysis `evolution`, and `conformance` companion packages. Treat its API as experimental until it reaches v1. The `telemetry` interface and its `slog`, `otel`, and `datadog` adapters are released. The `sink` egress seam and its destination -adapters, and the `source` ingress seam with its Kafka and JetStream adapters, -CloudEvents codec, reliability middleware, and state-machine bridge, are now -available and documented; the `broker` module is planned. +adapters, and the `source` ingress seam with its Kafka, JetStream, and Redis +Streams adapters, CloudEvents codec, reliability middleware, and state-machine +bridge, are now available and documented; the `broker` module is planned. ## Roadmap: event-driven seams diff --git a/docs/src/content/docs/source/adapters.md b/docs/src/content/docs/source/adapters.md index ead8748..1f5e6f4 100644 --- a/docs/src/content/docs/source/adapters.md +++ b/docs/src/content/docs/source/adapters.md @@ -45,6 +45,25 @@ deliver-by-start options. A JetStream durable consumer is the grouping analog of a Kafka consumer group, but it has no partitions and no assignment callbacks, so JetStream does **not** pretend to be a `ConsumerGroups` backend. +## redis (go-redis) + +`crucible/source/redis` is a consumer-group reader over go-redis: + +```go +in, _ := redis.New(redis.WithAddr("localhost:6379"), redis.WithGroup("orders-svc"), redis.WithConsumer("worker-1")) +sub, _ := in.Subscribe(ctx, source.SubscribeConfig{Topics: []string{"orders"}}) +m, _ := sub.Next(ctx) +_ = sub.Settle(ctx, m, source.Ack()) +``` + +It reads a Redis Stream with `XREADGROUP`, acks with `XACK`, leaves a naked +entry in the pending list for a later `XPENDING` plus `XCLAIM` redelivery, and +routes a terminated entry to a configured dead-letter stream before acking the +original. Replay is by entry ID through `XRANGE`, and lag comes from +`XINFO GROUPS` with an `XLEN` fallback. A Redis consumer group is the grouping +analog of a Kafka consumer group, but it has no partitions and no assignment +callbacks, so Redis does **not** pretend to be a `ConsumerGroups` backend. + ## Capability table per backend Capabilities are detected by interface assertion, once, inside the engine. The @@ -52,17 +71,17 @@ table is honest: an adapter satisfies a capability only when its backend truly supports it, and a compile-time `var _ Seekable = ...` assertion in each module keeps it accurate. -| Capability | kafka | jetstream | Notes | -|---|---|---|---| -| `Seekable` | yes | yes | live `SetOffsets` on Kafka; consumer-recreate on JetStream | -| `ConsumerGroups` | yes | no | Kafka rebalance hooks; JetStream has no partition assignment | -| `SharedDurable` | no | yes | the JetStream durable-consumer grouping analog | -| `PartitionOrdered` | yes | no | Kafka per-partition order | -| `OrderedDelivery` | no | yes | JetStream `OrderedConsumer`, single-threaded | -| `Batched` | yes | yes | batched fetch on both | -| `Transactional` | yes | no | Kafka EOS only; JetStream does not, and does not fake it | -| `Deduper` | yes | yes | dedup seam (see [reliability](/crucible/source/reliability/)) | -| `LagReporter` | yes | yes | consumer-lag gauge | +| Capability | kafka | jetstream | redis | Notes | +|---|---|---|---|---| +| `Seekable` | yes | yes | yes | live `SetOffsets` on Kafka; consumer-recreate on JetStream; entry-ID `XRANGE` on Redis | +| `ConsumerGroups` | yes | no | no | Kafka rebalance hooks; JetStream and Redis have no partition assignment | +| `SharedDurable` | no | yes | yes | the JetStream durable-consumer and Redis consumer-group grouping analogs | +| `PartitionOrdered` | yes | no | no | Kafka per-partition order | +| `OrderedDelivery` | no | yes | no | JetStream `OrderedConsumer`, single-threaded | +| `Batched` | yes | yes | no | batched fetch on Kafka and JetStream | +| `Transactional` | yes | no | no | Kafka EOS only; JetStream and Redis do not, and do not fake it | +| `Deduper` | yes | yes | no | dedup seam (see [reliability](/crucible/source/reliability/)) | +| `LagReporter` | yes | yes | yes | consumer-lag gauge; Redis reports group lag from `XINFO GROUPS` | The divergences are documented, never papered over. A `Nak(delay)` is a real delayed redelivery on JetStream but is best-effort on Kafka (pause plus reseek), @@ -70,8 +89,8 @@ and that is called out where it matters. ## Roadmap and bring your own -`source/redis` (Redis Streams) and `source/cdc` (Debezium/OpenCDC -change-data-capture) are on the roadmap. The catalog is a convenience: any type +`source/cdc` (Debezium/OpenCDC change-data-capture) is on the roadmap. The +catalog is a convenience: any type that satisfies the `Inlet` interface is an inlet, and the [`memsource`](/crucible/source/reliability/#testing-the-loop) in-memory adapter is itself an `Inlet`, so you can drive the whole engine in a unit test with no diff --git a/magefiles/magefile.go b/magefiles/magefile.go index b447e32..8b0e6c9 100644 --- a/magefiles/magefile.go +++ b/magefiles/magefile.go @@ -56,12 +56,12 @@ var sinkDestinations = []string{ // sourceDestinations are the standalone source modules kept out of the workspace // (see modules above): the SDK-backed sources (source/kafka via franz-go, -// source/jetstream via nats.go, source/cloudevents via the CloudEvents SDK) plus -// the flagship examples/sourcedrive. Each builds via its own go.mod and replace -// directives, so the SourceDestinations and Integration targets run with -// GOWORK=off. +// source/jetstream via nats.go, source/redis via go-redis, source/cloudevents +// via the CloudEvents SDK) plus the flagship examples/sourcedrive. Each builds +// via its own go.mod and replace directives, so the SourceDestinations and +// Integration targets run with GOWORK=off. var sourceDestinations = []string{ - "source/kafka", "source/jetstream", "source/cloudevents", "examples/sourcedrive", + "source/kafka", "source/jetstream", "source/redis", "source/cloudevents", "examples/sourcedrive", } // Pinned tool versions — keep in sync with .github/workflows/ci.yml. diff --git a/source/redis/README.md b/source/redis/README.md new file mode 100644 index 0000000..44a48cf --- /dev/null +++ b/source/redis/README.md @@ -0,0 +1,79 @@ +# source/redis + +A [`crucible/source`](../) ingress adapter that consumes a Redis Stream through +a consumer group. Runtime dependency: +[`github.com/redis/go-redis/v9`](https://github.com/redis/go-redis). + +```go +in, err := redis.New( + redis.WithAddr("localhost:6379"), + redis.WithGroup("orders-svc"), + redis.WithConsumer("worker-1"), + redis.WithDLQStream("orders.dlq"), + redis.WithBlock(5*time.Second), + redis.WithCount(64), +) +sub, _ := in.Subscribe(ctx, source.SubscribeConfig{Topics: []string{"orders"}}) + +m, _ := sub.Next(ctx) // or drive sub with a source.Hopper +_ = sub.Settle(ctx, m, source.Ack()) +``` + +Construct an `Inlet` with `New` and functional options (`WithAddr`/`WithClient`, +`WithGroup`, `WithConsumer`, `WithDLQStream`, `WithBlock`, `WithCount`, +`WithMinIdle`). The returned `Subscription` reads entries with `Next` (via +`XREADGROUP`) and applies a handler `source.Result` with `Settle`. Backpressure +comes from the `Count` batch size plus the block window. Reach the underlying +`*redis.Client` through `Inlet.As`, or the per-entry `redis.XMessage` through +`Message.As`; no vendor type appears in the adapter's own signatures. + +A stream entry maps onto `source.Message` as follows: the `value` field becomes +the raw `Value`, the `crucible-key` field (when set) becomes the routing `Key` +(otherwise the stream name), every field is exposed as a `Header`, `Subject` is +the stream, and `Cursor` is the entry ID. + +## Settle vocabulary + +A consumer group reads with `XREADGROUP`; every delivered entry stays in the +group's Pending Entries List (PEL) until it is settled. + +- **Ack** calls `XACK`, removing the entry from the PEL. +- **Nak** leaves the entry in the PEL. A consumer reclaims and redelivers it by + scanning the backlog with `XPENDING` + `XCLAIM` once it has idled past the + minimum; call `NakRedeliver` (on a timer or between read cycles) to drive that + pass. The cadence is a deployment choice, so the engine does not force it. +- **Term** (and **Reject**) append the entry's fields plus dead-letter metadata + (`crucible-dlq-original-id`, `crucible-dlq-stream`, `crucible-dlq-class`, + `crucible-dlq-error`) to the configured `WithDLQStream`, then `XACK` the + original. With no dead-letter stream configured, a terminated entry is acked + and dropped. +- **InProgress** is a no-op: Redis has no per-message deadline to extend. +- **Manual** is a no-op; the handler settled the entry itself through + `Message.As` and the client. + +## Capabilities + +The subscription honestly advertises the Redis-shaped capabilities by type +assertion: `source.SharedDurable` (the consumer group is the competing-consumer +analog), `source.Seekable` (replay by entry ID via `XRANGE`, with `SeekToTime` +translating a timestamp into an entry ID), and `source.LagReporter` (group lag +from `XINFO GROUPS`, falling back to `XLEN`). + +## Redis divergences from the source contract + +- **No `ConsumerGroups`.** A Redis consumer group has no partitions and no + assignment lifecycle, so the adapter does not implement + `source.ConsumerGroups`. The group load-balances across processes instead, + surfaced as `source.SharedDurable`. `PartitionKey` is always `""`, so the + Hopper shards by `Key`. +- **No `Transactional`.** Redis Streams offer no consume-side transaction, so + the adapter does not implement `source.Transactional`; the capability is + absent rather than faked. + +## Stability + +Experimental (pre-v1). The API may change until the suite locks v1.0.0. + +## License + +Apache-2.0. See [LICENSE](../../LICENSE) and [NOTICE](../../NOTICE). diff --git a/source/redis/example_test.go b/source/redis/example_test.go new file mode 100644 index 0000000..9e72f25 --- /dev/null +++ b/source/redis/example_test.go @@ -0,0 +1,87 @@ +// SPDX-License-Identifier: Apache-2.0 + +package redis_test + +import ( + "context" + "fmt" + + goredis "github.com/redis/go-redis/v9" + + "github.com/stablekernel/crucible/source" + credis "github.com/stablekernel/crucible/source/redis" +) + +// exampleClient is a stand-in Client seam that yields one stream entry, so the +// example runs without a live Redis server. It implements only the methods the +// consume-and-ack path touches; a real program passes redis.WithAddr or +// redis.WithClient(realClient) instead. +type exampleClient struct { + credis.Client + delivered bool +} + +func (c *exampleClient) XGroupCreateMkStream(ctx context.Context, _, _, _ string) *goredis.StatusCmd { + return goredis.NewStatusCmd(ctx) +} + +func (c *exampleClient) XReadGroup(ctx context.Context, a *goredis.XReadGroupArgs) *goredis.XStreamSliceCmd { + cmd := goredis.NewXStreamSliceCmd(ctx) + if c.delivered { + cmd.SetErr(goredis.Nil) // no further entries + return cmd + } + c.delivered = true + cmd.SetVal([]goredis.XStream{{ + Stream: a.Streams[0], + Messages: []goredis.XMessage{{ + ID: "1526919030474-0", + Values: map[string]any{"value": "placed", "crucible-key": "A-1"}, + }}, + }}) + return cmd +} + +func (c *exampleClient) XAck(ctx context.Context, _, _ string, _ ...string) *goredis.IntCmd { + cmd := goredis.NewIntCmd(ctx) + cmd.SetVal(1) + return cmd +} + +// Example shows the consume-and-settle loop: open a consumer-group +// subscription, take the next entry off the stream, and ack it after handling. +// In a real program the seam is a live client supplied via WithAddr or +// WithClient. +func Example() { + in, err := credis.New( + credis.WithClient(&exampleClient{}), + credis.WithGroup("orders-svc"), + credis.WithConsumer("worker-1"), + ) + if err != nil { + panic(err) + } + defer func() { _ = in.Close() }() + + sub, err := in.Subscribe(context.Background(), source.SubscribeConfig{Topics: []string{"orders"}}) + if err != nil { + panic(err) + } + defer func() { _ = sub.Close() }() + + m, err := sub.Next(context.Background()) + if err != nil { + panic(err) + } + if err := sub.Settle(context.Background(), m, source.Ack()); err != nil { + panic(err) + } + + fmt.Println(m.Subject()) + fmt.Println(string(m.Key())) + fmt.Println(string(m.Value())) + // Output: + // orders + // A-1 + // placed +} diff --git a/source/redis/go.mod b/source/redis/go.mod new file mode 100644 index 0000000..f1c5b5a --- /dev/null +++ b/source/redis/go.mod @@ -0,0 +1,70 @@ +module github.com/stablekernel/crucible/source/redis + +go 1.25.11 + +replace github.com/stablekernel/crucible/source => ../ + +replace github.com/stablekernel/crucible/telemetry => ../../telemetry + +require ( + github.com/redis/go-redis/v9 v9.20.0 + github.com/stablekernel/crucible/source v0.0.0-00010101000000-000000000000 + github.com/testcontainers/testcontainers-go v0.42.0 + github.com/testcontainers/testcontainers-go/modules/redis v0.42.0 +) + +require ( + dario.cat/mergo v1.0.2 // indirect + github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect + github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/containerd/errdefs v1.0.0 // indirect + github.com/containerd/errdefs/pkg v0.3.0 // indirect + github.com/containerd/log v0.1.0 // indirect + github.com/containerd/platforms v0.2.1 // indirect + github.com/cpuguy83/dockercfg v0.3.2 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/distribution/reference v0.6.0 // indirect + github.com/docker/go-connections v0.6.0 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/ebitengine/purego v0.10.0 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-ole/go-ole v1.2.6 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/klauspost/compress v1.18.5 // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/magiconair/properties v1.8.10 // indirect + github.com/mdelapenya/tlscert v0.2.0 // indirect + github.com/moby/docker-image-spec v1.3.1 // indirect + github.com/moby/go-archive v0.2.0 // indirect + github.com/moby/moby/api v1.54.1 // indirect + github.com/moby/moby/client v0.4.0 // indirect + github.com/moby/patternmatcher v0.6.1 // indirect + github.com/moby/sys/sequential v0.6.0 // indirect + github.com/moby/sys/user v0.4.0 // indirect + github.com/moby/sys/userns v0.1.0 // indirect + github.com/moby/term v0.5.2 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect + github.com/shirou/gopsutil/v4 v4.26.3 // indirect + github.com/sirupsen/logrus v1.9.4 // indirect + github.com/stablekernel/crucible/telemetry v0.0.0 // indirect + github.com/stretchr/testify v1.11.1 // indirect + github.com/tklauser/go-sysconf v0.3.16 // indirect + github.com/tklauser/numcpus v0.11.0 // 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.60.0 // indirect + go.opentelemetry.io/otel v1.41.0 // indirect + go.opentelemetry.io/otel/metric v1.41.0 // indirect + go.opentelemetry.io/otel/trace v1.41.0 // indirect + go.uber.org/atomic v1.11.0 // indirect + golang.org/x/crypto v0.48.0 // indirect + golang.org/x/sys v0.42.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/source/redis/go.sum b/source/redis/go.sum new file mode 100644 index 0000000..eee7d55 --- /dev/null +++ b/source/redis/go.sum @@ -0,0 +1,151 @@ +dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= +dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= +github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= +github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= +github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= +github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= +github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= +github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A= +github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw= +github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA= +github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= +github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= +github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= +github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pMmjSD94= +github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE= +github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= +github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU= +github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= +github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= +github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= +github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= +github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= +github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/mdelapenya/tlscert v0.2.0 h1:7H81W6Z/4weDvZBNOfQte5GpIMo0lGYEeWbkGp5LJHI= +github.com/mdelapenya/tlscert v0.2.0/go.mod h1:O4njj3ELLnJjGdkN7M/vIVCpZ+Cf0L6muqOG4tLSl8o= +github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= +github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= +github.com/moby/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8= +github.com/moby/go-archive v0.2.0/go.mod h1:mNeivT14o8xU+5q1YnNrkQVpK+dnNe/K6fHqnTg4qPU= +github.com/moby/moby/api v1.54.1 h1:TqVzuJkOLsgLDDwNLmYqACUuTehOHRGKiPhvH8V3Nn4= +github.com/moby/moby/api v1.54.1/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs= +github.com/moby/moby/client v0.4.0 h1:S+2XegzHQrrvTCvF6s5HFzcrywWQmuVnhOXe2kiWjIw= +github.com/moby/moby/client v0.4.0/go.mod h1:QWPbvWchQbxBNdaLSpoKpCdf5E+WxFAgNHogCWDoa7g= +github.com/moby/patternmatcher v0.6.1 h1:qlhtafmr6kgMIJjKJMDmMWq7WLkKIo23hsrpR3x084U= +github.com/moby/patternmatcher v0.6.1/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= +github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU= +github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko= +github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= +github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs= +github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g= +github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28= +github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= +github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= +github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0= +github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/shirou/gopsutil/v4 v4.26.3 h1:2ESdQt90yU3oXF/CdOlRCJxrP+Am1aBYubTMTfxJ1qc= +github.com/shirou/gopsutil/v4 v4.26.3/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/testcontainers/testcontainers-go v0.42.0 h1:He3IhTzTZOygSXLJPMX7n44XtK+qhjat1nI9cneBbUY= +github.com/testcontainers/testcontainers-go v0.42.0/go.mod h1:vZjdY1YmUA1qEForxOIOazfsrdyORJAbhi0bp8plN30= +github.com/testcontainers/testcontainers-go/modules/redis v0.42.0 h1:id/6LH8ZeDrtAUVSuNvZUAJ1kVpb82y1pr9yweAWsRg= +github.com/testcontainers/testcontainers-go/modules/redis v0.42.0/go.mod h1:uF0jI8FITagQpBNOgweGBmPf6rP4K0SeL1XFPbsZSSY= +github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA= +github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI= +github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw= +github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ= +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= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ= +go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= +go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= +go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= +go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= +go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY= +go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg= +go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o= +go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w= +go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= +go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= +golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= +golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.40.0 h1:36e4zGLqU4yhjlmxEaagx2KuYbJq3EwY8K943ZsHcvg= +golang.org/x/term v0.40.0/go.mod h1:w2P8uVp06p2iyKKuvXIm7N/y0UCRt3UfJTfZ7oOpglM= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= +gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= +pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= +pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04= diff --git a/source/redis/integration_test.go b/source/redis/integration_test.go new file mode 100644 index 0000000..ac341e0 --- /dev/null +++ b/source/redis/integration_test.go @@ -0,0 +1,176 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build integration + +package redis_test + +import ( + "context" + "errors" + "testing" + "time" + + goredis "github.com/redis/go-redis/v9" + "github.com/testcontainers/testcontainers-go" + tcredis "github.com/testcontainers/testcontainers-go/modules/redis" + + "github.com/stablekernel/crucible/source" + credis "github.com/stablekernel/crucible/source/redis" +) + +// TestIntegrationConsumeAckNakTerm starts a real Redis container, XADDs three +// entries to a stream, consumes them through the Inlet's consumer group, and +// exercises the settle vocabulary against a live server: +// +// - the ack'd entry is XACK'd off the pending list, +// - the nak'd entry stays pending and is redelivered via NakRedeliver, then +// ack'd, +// - the term'd entry lands on the dead-letter stream and is ack'd off the +// original. +// +// It skips cleanly when Docker is not reachable. +func TestIntegrationConsumeAckNakTerm(t *testing.T) { + t.Parallel() + skipWithoutDocker(t) + + ctx := context.Background() + container, err := tcredis.Run(ctx, "redis:7.4-alpine") + if err != nil { + t.Skipf("redis.Run() unavailable (image pull or startup failed); skipping: %v", err) + } + t.Cleanup(func() { _ = testcontainers.TerminateContainer(container) }) + + endpoint, err := container.ConnectionString(ctx) + if err != nil { + t.Fatalf("ConnectionString() error = %v", err) + } + opts, err := goredis.ParseURL(endpoint) + if err != nil { + t.Fatalf("ParseURL() error = %v", err) + } + client := goredis.NewClient(opts) + t.Cleanup(func() { _ = client.Close() }) + + const ( + stream = "orders" + dlq = "orders.dlq" + ) + // Produce three entries: one to ack, one to nak-then-ack, one to term. + for _, body := range []string{"ack-me", "nak-me", "term-me"} { + if err = client.XAdd(ctx, &goredis.XAddArgs{ + Stream: stream, + ID: "*", + Values: map[string]any{"value": body}, + }).Err(); err != nil { + t.Fatalf("XAdd(%q) error = %v", body, err) + } + } + + in, err := credis.New( + credis.WithClient(client), + credis.WithGroup("it-workers"), + credis.WithConsumer("worker-1"), + credis.WithDLQStream(dlq), + credis.WithBlock(time.Second), + credis.WithCount(16), + credis.WithMinIdle(50*time.Millisecond), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + t.Cleanup(func() { _ = in.Close() }) + + sub, err := in.Subscribe(ctx, source.SubscribeConfig{Topics: []string{stream}}) + if err != nil { + t.Fatalf("Subscribe() error = %v", err) + } + t.Cleanup(func() { _ = sub.Close() }) + + // First pass: ack one, term one, nak one (leaving it pending). + var nakedID string + for seen := 0; seen < 3; { + nctx, cancel := context.WithTimeout(ctx, 3*time.Second) + m, nerr := sub.Next(nctx) + cancel() + if nerr != nil { + if errors.Is(nerr, context.DeadlineExceeded) { + continue + } + t.Fatalf("Next() error = %v", nerr) + } + seen++ + switch string(m.Value()) { + case "ack-me": + if err := sub.Settle(ctx, m, source.Ack()); err != nil { + t.Fatalf("Settle(ack) error = %v", err) + } + case "term-me": + if err := sub.Settle(ctx, m, source.Term(errors.New("poison"))); err != nil { + t.Fatalf("Settle(term) error = %v", err) + } + case "nak-me": + nakedID = m.Cursor().String() + if err := sub.Settle(ctx, m, source.Nak(errors.New("transient"))); err != nil { + t.Fatalf("Settle(nak) error = %v", err) + } + } + } + + // The term'd entry must have landed on the dead-letter stream. + dlqEntries, err := client.XRange(ctx, dlq, "-", "+").Result() + if err != nil { + t.Fatalf("XRange(dlq) error = %v", err) + } + if len(dlqEntries) != 1 || dlqEntries[0].Values["value"] != "term-me" { + t.Fatalf("dlq entries = %#v, want one term-me", dlqEntries) + } + + // The nak'd entry stays pending: redeliver it after it idles past MinIdle. + // NakRedeliver is a Redis-specific surface beyond the source.Subscription + // interface, reached by asserting the concrete capability. + redeliverer, ok := sub.(interface { + NakRedeliver(ctx context.Context, minIdle time.Duration) (int, error) + }) + if !ok { + t.Fatal("subscription does not expose NakRedeliver") + } + time.Sleep(100 * time.Millisecond) + n, err := redeliverer.NakRedeliver(ctx, 50*time.Millisecond) + if err != nil { + t.Fatalf("NakRedeliver() error = %v", err) + } + if n != 1 { + t.Fatalf("NakRedeliver() reclaimed %d, want 1", n) + } + m, err := sub.Next(ctx) + if err != nil { + t.Fatalf("Next() after redeliver error = %v", err) + } + if string(m.Value()) != "nak-me" || m.Cursor().String() != nakedID { + t.Fatalf("redelivered = %q (%s), want nak-me (%s)", m.Value(), m.Cursor(), nakedID) + } + if err := sub.Settle(ctx, m, source.Ack()); err != nil { + t.Fatalf("Settle(ack after nak) error = %v", err) + } + + // Nothing remains pending once every entry is settled. + pending, err := client.XPending(ctx, stream, "it-workers").Result() + if err != nil { + t.Fatalf("XPending() error = %v", err) + } + if pending.Count != 0 { + t.Fatalf("pending count = %d, want 0 after all settled", pending.Count) + } +} + +func skipWithoutDocker(t *testing.T) { + t.Helper() + provider, err := testcontainers.NewDockerProvider() + if err != nil { + t.Skipf("docker unavailable: %v", err) + } + defer func() { _ = provider.Close() }() + if err := provider.Health(context.Background()); err != nil { + t.Skipf("docker unavailable: %v", err) + } +} diff --git a/source/redis/message.go b/source/redis/message.go new file mode 100644 index 0000000..e9eb193 --- /dev/null +++ b/source/redis/message.go @@ -0,0 +1,131 @@ +// SPDX-License-Identifier: Apache-2.0 + +package redis + +import ( + goredis "github.com/redis/go-redis/v9" + + "github.com/stablekernel/crucible/source" +) + +// KeyHeader is the entry field an inbound stream entry may set to carry an +// explicit partitioning/routing key. When present, the adapter uses its value +// as [source.Message.Key]; otherwise the key falls back to the stream name so +// the Hopper still shards deterministically. A Redis Stream has no partitions, +// so [source.Message.PartitionKey] is always "" and the Hopper shards by Key. +const KeyHeader = "crucible-key" + +// ValueField is the entry field the adapter reads as the raw payload bytes when +// an entry carries one. A producer that writes a single payload field under +// this name round-trips cleanly: the field becomes [source.Message.Value] and +// every other field becomes a header. When the field is absent, Value is nil +// and the entry's fields are still exposed as headers. +const ValueField = "value" + +// idCursor is the resumable position of a Redis Stream entry: its entry ID +// ("millisecondsTime-sequence"). It satisfies [source.Cursor] and is the value +// a [subscription.SeekToCursor] resumes from via XRANGE. +type idCursor string + +// String renders the entry ID for logs and diagnostics. +func (c idCursor) String() string { return string(c) } + +// message adapts a go-redis XMessage onto [source.Message]. The vendor entry is +// reachable only through As; the neutral accessors expose the common surface. +// The fields are snapshotted at construction so the neutral view is a stable +// value independent of later driver state. +type message struct { + entry goredis.XMessage + stream string + headers source.Headers + key []byte + value []byte +} + +// newMessage wraps a go-redis XMessage from stream, projecting its fields onto +// the neutral surface: the [ValueField] field (when present) becomes the raw +// value, the [KeyHeader] field (when present) becomes the routing key, and every +// field is exposed as a header. The key falls back to the stream name so the +// Hopper always has a deterministic shard key on a backend with no partitions. +func newMessage(stream string, entry goredis.XMessage) *message { + headers := make(source.Headers, 0, len(entry.Values)) + var value []byte + for k, v := range entry.Values { + s := toString(v) + headers = append(headers, source.Header{Key: k, Value: s}) + if k == ValueField { + value = []byte(s) + } + } + + key := []byte(stream) + if v, ok := entry.Values[KeyHeader]; ok { + if s := toString(v); s != "" { + key = []byte(s) + } + } + + return &message{ + entry: entry, + stream: stream, + headers: headers, + key: key, + value: value, + } +} + +// toString renders a Redis field value as a string. Redis stream fields are +// always strings on the wire; go-redis decodes them as string, so the type +// switch is defensive and the string branch is the live path. +func toString(v any) string { + if s, ok := v.(string); ok { + return s + } + if v == nil { + return "" + } + if s, ok := v.([]byte); ok { + return string(s) + } + return "" +} + +// Key returns the routing key: the [KeyHeader] field value when set, otherwise +// the stream name, so the Hopper always has a deterministic shard key on a +// backend with no partitions. +func (m *message) Key() []byte { return m.key } + +// Value returns the raw payload bytes: the [ValueField] field when present, else +// nil. The full set of entry fields remains available through [message.Headers]. +func (m *message) Value() []byte { return m.value } + +// Headers returns the entry's fields as a value-type slice. +func (m *message) Headers() source.Headers { return m.headers } + +// Subject returns the stream the entry arrived on. +func (m *message) Subject() string { return m.stream } + +// PartitionKey returns "" because a Redis Stream has no partitions; the Hopper +// shards by Key instead. +func (m *message) PartitionKey() string { return "" } + +// Cursor returns the entry ID as a resumable [source.Cursor]. +func (m *message) Cursor() source.Cursor { return idCursor(m.entry.ID) } + +// As assigns the underlying go-redis XMessage to target if target is a +// *redis.XMessage, returning whether it did. It is the escape hatch to reach the +// vendor entry (for the raw fields, say) without a vendor type in the neutral +// surface. +func (m *message) As(target any) bool { + if p, ok := target.(*goredis.XMessage); ok { + *p = m.entry + return true + } + return false +} + +// compile-time assertions. +var ( + _ source.Message = (*message)(nil) + _ source.Cursor = idCursor("") +) diff --git a/source/redis/redis.go b/source/redis/redis.go new file mode 100644 index 0000000..2d9c5f6 --- /dev/null +++ b/source/redis/redis.go @@ -0,0 +1,653 @@ +// SPDX-License-Identifier: Apache-2.0 + +// Package redis is a source ingress adapter that consumes a Redis Stream through +// a consumer group. It wraps the stream surface of +// [github.com/redis/go-redis/v9] behind a narrow seam interface so the real +// *redis.Client satisfies it structurally while unit tests drive a hand-rolled +// fake with no live server. +// +// Construct an [Inlet] with [New] and functional options, then drive its +// [github.com/stablekernel/crucible/source.Subscription] with a source.Hopper +// (or read it directly via Next/Settle). +// +// in, err := redis.New( +// redis.WithAddr("localhost:6379"), +// redis.WithGroup("orders-svc"), +// redis.WithConsumer("worker-1"), +// redis.WithDLQStream("orders.dlq"), +// redis.WithBlock(5*time.Second), +// redis.WithCount(64), +// ) +// sub, err := in.Subscribe(ctx, source.SubscribeConfig{Topics: []string{"orders"}}) +// m, err := sub.Next(ctx) // or drive with a source.Hopper +// err = sub.Settle(ctx, m, source.Ack()) +// +// # Settle vocabulary on Redis Streams +// +// A consumer group reads with XREADGROUP; every delivered entry stays in the +// group's Pending Entries List (PEL) until it is settled. +// +// - ActionAck calls XACK, removing the entry from the PEL. +// - ActionNak leaves the entry in the PEL: it is redelivered to a consumer that +// scans the pending backlog with XPENDING + XCLAIM after the configured min +// idle time. Result.Requeue raises the claim's min-idle floor when larger. +// - ActionTerm appends the entry (with its original fields plus dead-letter +// metadata) to the configured dead-letter stream, then XACKs the original so +// it is not redelivered. +// - ActionInProgress is a no-op: Redis has no per-message deadline to extend. +// - ActionManual is a no-op; the handler settled the entry itself through +// Message.As and the client. +// +// # Redis divergences from the source contract +// +// A Redis consumer group has no partitions and no assignment lifecycle, so this +// adapter does NOT implement +// [github.com/stablekernel/crucible/source.ConsumerGroups]. The group is the +// competing-consumer analog instead, surfaced through +// [github.com/stablekernel/crucible/source.SharedDurable]: multiple consumers in +// one group load-balance the stream without partition assignment. +// +// Redis Streams offer no consume-side transaction, so this adapter does NOT +// implement [github.com/stablekernel/crucible/source.Transactional]; the +// capability is absent rather than faked. +// +// # Stability +// +// Experimental (pre-v1); the API may change until the suite locks v1.0.0. +package redis + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + goredis "github.com/redis/go-redis/v9" + + "github.com/stablekernel/crucible/source" +) + +// defaultBlock is the XREADGROUP block duration used when none is configured: a +// bounded wait keeps Next responsive to context cancellation without busy- +// looping when the stream is idle. +const defaultBlock = 5 * time.Second + +// defaultCount is the XREADGROUP batch size used when none is configured. +const defaultCount = 64 + +// defaultMinIdle is the floor a pending entry must be idle before this consumer +// claims it for redelivery, used when WithBlock does not imply a larger value. +const defaultMinIdle = 30 * time.Second + +// Client is the narrow Redis Streams surface this adapter needs. It is satisfied +// structurally by *redis.Client from github.com/redis/go-redis/v9, so callers +// wire the real client while unit tests substitute a fake without a live server. +type Client interface { + // XGroupCreateMkStream creates the consumer group, creating the stream if it + // does not yet exist. + XGroupCreateMkStream(ctx context.Context, stream, group, start string) *goredis.StatusCmd + // XReadGroup reads new or pending entries for a consumer group. + XReadGroup(ctx context.Context, a *goredis.XReadGroupArgs) *goredis.XStreamSliceCmd + // XAck acknowledges entries, removing them from the group's pending list. + XAck(ctx context.Context, stream, group string, ids ...string) *goredis.IntCmd + // XPendingExt lists pending entries with their idle time and delivery count. + XPendingExt(ctx context.Context, a *goredis.XPendingExtArgs) *goredis.XPendingExtCmd + // XClaim reassigns ownership of idle pending entries to this consumer. + XClaim(ctx context.Context, a *goredis.XClaimArgs) *goredis.XMessageSliceCmd + // XAdd appends an entry to a stream (used for dead-lettering). + XAdd(ctx context.Context, a *goredis.XAddArgs) *goredis.StringCmd + // XRange returns entries in an ID range (used for cursor seeks). + XRange(ctx context.Context, stream, start, stop string) *goredis.XMessageSliceCmd + // XLen returns the number of entries in a stream. + XLen(ctx context.Context, stream string) *goredis.IntCmd + // XInfoGroups returns the consumer-group metadata for a stream, including each + // group's lag behind the tail. + XInfoGroups(ctx context.Context, key string) *goredis.XInfoGroupsCmd +} + +// compile-time assertion that the real go-redis client satisfies the seam. +var _ Client = (*goredis.Client)(nil) + +// Inlet is a Redis Streams ingress adapter. It opens consumer-group +// subscriptions onto a configured stream. Build one with [New]; it is safe for +// concurrent use. +type Inlet struct { + client Client + ownConn bool // true when New dialed the client and Close must shut it down + cfg config +} + +// config holds the resolved construction options for an [Inlet]. +type config struct { + addr string + client Client + group string + consumer string + dlqStream string + block time.Duration + count int64 + minIdle time.Duration +} + +// Option configures an [Inlet] at construction. Options compose; later options +// override earlier ones for the same field. +type Option func(*config) + +// WithAddr dials a Redis client at addr (host:port). Mutually exclusive with +// [WithClient]: supply exactly one. When WithAddr is used the [Inlet] owns the +// client and closes it on [Inlet.Close]. +func WithAddr(addr string) Option { return func(c *config) { c.addr = addr } } + +// WithClient uses an existing client (anything satisfying [Client], such as a +// *redis.Client). The caller retains ownership; the [Inlet] does not close a +// client it did not dial. Mutually exclusive with [WithAddr]. +func WithClient(client Client) Option { return func(c *config) { c.client = client } } + +// WithGroup sets the consumer group name: the competing-consumer grouping that +// satisfies [source.SharedDurable]. Required. +func WithGroup(group string) Option { return func(c *config) { c.group = group } } + +// WithConsumer sets this consumer's name within the group. Each process in a +// group needs a distinct consumer name so the group can track per-consumer +// pending entries. Required. +func WithConsumer(name string) Option { return func(c *config) { c.consumer = name } } + +// WithDLQStream sets the dead-letter stream an [source.ActionTerm] (Term/Reject) +// routes a rejected entry to before acking the original. Empty disables dead- +// lettering: a terminated entry is acked and dropped. +func WithDLQStream(stream string) Option { return func(c *config) { c.dlqStream = stream } } + +// WithBlock sets how long XREADGROUP blocks waiting for a new entry before +// returning empty so Next can re-check its context. Zero leaves the default. +func WithBlock(d time.Duration) Option { return func(c *config) { c.block = d } } + +// WithCount sets the maximum number of entries XREADGROUP fetches per call: the +// fetch-side batch bound. Zero leaves the default. +func WithCount(n int64) Option { return func(c *config) { c.count = n } } + +// WithMinIdle sets how long a pending entry must be idle before this consumer +// claims it for redelivery. It is the redelivery floor for naked entries; a +// larger [source.Result.Requeue] on a Nak raises it per entry. Zero leaves the +// default. +func WithMinIdle(d time.Duration) Option { return func(c *config) { c.minIdle = d } } + +// New builds an [Inlet] from opts. Exactly one of [WithAddr] or [WithClient] +// must be supplied, and [WithGroup] and [WithConsumer] are required. When +// [WithAddr] is used New dials the client eagerly so misconfiguration surfaces +// here rather than at first Subscribe. +func New(opts ...Option) (*Inlet, error) { + var cfg config + for _, opt := range opts { + opt(&cfg) + } + if cfg.group == "" { + return nil, fmt.Errorf("redis: WithGroup is required") + } + if cfg.consumer == "" { + return nil, fmt.Errorf("redis: WithConsumer is required") + } + if cfg.block == 0 { + cfg.block = defaultBlock + } + if cfg.count == 0 { + cfg.count = defaultCount + } + if cfg.minIdle == 0 { + cfg.minIdle = defaultMinIdle + } + + client := cfg.client + ownConn := false + switch { + case client != nil && cfg.addr != "": + return nil, fmt.Errorf("redis: WithAddr and WithClient are mutually exclusive") + case client != nil: + // caller-owned client + case cfg.addr != "": + client = goredis.NewClient(&goredis.Options{Addr: cfg.addr}) + ownConn = true + default: + return nil, fmt.Errorf("redis: one of WithAddr or WithClient is required") + } + + return &Inlet{client: client, ownConn: ownConn, cfg: cfg}, nil +} + +// Subscribe opens a consumer-group Subscription. cfg.Topics must name exactly +// one stream (Redis Streams are single-stream per group read); cfg.Group, when +// set, overrides [WithGroup]. The consumer group is created lazily on the first +// [subscription.Next] (with MKSTREAM, so a not-yet-existent stream is created), +// keeping Subscribe cheap. +func (in *Inlet) Subscribe(_ context.Context, cfg source.SubscribeConfig) (source.Subscription, error) { + if len(cfg.Topics) != 1 { + return nil, fmt.Errorf("redis: Subscribe requires exactly one stream in Topics, got %d", len(cfg.Topics)) + } + group := in.cfg.group + if cfg.Group != "" { + group = cfg.Group + } + return &subscription{ + client: in.client, + stream: cfg.Topics[0], + group: group, + consumer: in.cfg.consumer, + dlqStream: in.cfg.dlqStream, + block: in.cfg.block, + count: in.cfg.count, + minIdle: in.cfg.minIdle, + startID: "0", // create the group at the stream origin by default + readFrom: ">", // read new (never-delivered) entries by default + }, nil +} + +// Close releases the inlet's resources. It shuts down the Redis client only when +// the inlet dialed it (via [WithAddr]); a caller-supplied client is left open. +// Close live Subscriptions first. +func (in *Inlet) Close() error { + if in.ownConn { + if c, ok := in.client.(interface{ Close() error }); ok { + return c.Close() + } + } + return nil +} + +// As assigns the inlet's underlying Redis client to target if target is a +// *redis.Client (or the narrow *Client seam), returning whether it did. It is +// the escape hatch to reach the driver for operations outside this adapter's +// surface. +func (in *Inlet) As(target any) bool { + switch p := target.(type) { + case *Client: + *p = in.client + return true + case **goredis.Client: + if rc, ok := in.client.(*goredis.Client); ok { + *p = rc + return true + } + } + return false +} + +// errSeekClosed reports a seek attempted on a drained subscription. +var errSeekClosed = errors.New("redis: cannot seek a closed subscription") + +// subscription is a live Redis consumer-group read surface. It implements +// source.Subscription plus the SharedDurable, Seekable, and LagReporter +// capabilities. Next is single-consumer (the Hopper's fetch loop); Settle is +// safe to call concurrently because the underlying client commands are. +type subscription struct { + client Client + stream string + group string + consumer string + dlqStream string + block time.Duration + count int64 + minIdle time.Duration + + mu sync.Mutex + created bool // consumer group ensured + closed bool + startID string // group-create start ("0" origin, "$" tail, or an explicit ID) + pending []source.Message // buffered, fetched-but-not-yet-yielded entries + inflight int // entries delivered by Next, not yet settled + readFrom string // ">" for new entries; an ID to drain the backlog +} + +// ensureGroup creates the consumer group (with MKSTREAM) once, idempotently +// tolerating the BUSYGROUP error that means the group already exists. The caller +// holds s.mu. +func (s *subscription) ensureGroup(ctx context.Context) error { + if s.created { + return nil + } + err := s.client.XGroupCreateMkStream(ctx, s.stream, s.group, s.startID).Err() + if err != nil && !isBusyGroup(err) { + return fmt.Errorf("redis: create group %q on %q: %w", s.group, s.stream, err) + } + s.created = true + return nil +} + +// isBusyGroup reports whether err is the "consumer group already exists" error, +// which ensureGroup treats as success so group creation is idempotent. Redis +// returns it as a BUSYGROUP error reply; go-redis surfaces it as a plain error +// whose message carries the prefix, so a substring match is the reliable test. +func isBusyGroup(err error) bool { + if err == nil { + return false + } + msg := err.Error() + const want = "BUSYGROUP" + for i := 0; i+len(want) <= len(msg); i++ { + if msg[i:i+len(want)] == want { + return true + } + } + return false +} + +// Next returns the next entry, blocking up to the configured block duration on +// XREADGROUP. It drains any buffered batch first, then reads new entries; an +// empty read returns after the block window so the caller's context is honored. +// Once the subscription is closed and its buffer drained, Next returns +// [source.ErrDrained]. +func (s *subscription) Next(ctx context.Context) (source.Message, error) { + for { + if err := ctx.Err(); err != nil { + return nil, err + } + s.mu.Lock() + if s.closed && len(s.pending) == 0 { + s.mu.Unlock() + return nil, source.ErrDrained + } + if len(s.pending) > 0 { + m := s.pending[0] + s.pending = s.pending[1:] + s.inflight++ + s.mu.Unlock() + return m, nil + } + if s.closed { + s.mu.Unlock() + return nil, source.ErrDrained + } + if err := s.ensureGroup(ctx); err != nil { + s.mu.Unlock() + return nil, err + } + readFrom := s.readFrom + s.mu.Unlock() + + batch, err := s.fetch(ctx, readFrom) + if err != nil { + return nil, err + } + s.mu.Lock() + // When draining the historical backlog (readFrom is an ID, not ">"), an + // empty result means the backlog is exhausted; switch to new entries. + if len(batch) == 0 && readFrom != ">" { + s.readFrom = ">" + s.mu.Unlock() + continue + } + s.pending = append(s.pending, batch...) + s.mu.Unlock() + // Loop: serve from the freshly filled buffer (or block again if empty). + } +} + +// fetch issues one XREADGROUP for readFrom (">" for new entries, an ID for the +// consumer's own pending backlog) and maps the returned entries onto neutral +// messages. A blocking read that times out with no entries returns an empty +// slice, not an error. +func (s *subscription) fetch(ctx context.Context, readFrom string) ([]source.Message, error) { + res, err := s.client.XReadGroup(ctx, &goredis.XReadGroupArgs{ + Group: s.group, + Consumer: s.consumer, + Streams: []string{s.stream, readFrom}, + Count: s.count, + Block: s.block, + }).Result() + if err != nil { + if errors.Is(err, goredis.Nil) { + return nil, nil // block window elapsed with no new entries + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return nil, ctx.Err() + } + return nil, fmt.Errorf("redis: read group: %w", err) + } + var out []source.Message + for _, st := range res { + for _, e := range st.Messages { + out = append(out, newMessage(s.stream, e)) + } + } + return out, nil +} + +// Settle applies the handler [source.Result] to m by translating its Action onto +// the Redis Streams settle vocabulary: ack (XACK), nak (leave pending for a +// later XCLAIM-driven redelivery), term (route to the dead-letter stream, then +// XACK), in-progress (no-op), or manual (no-op). +func (s *subscription) Settle(ctx context.Context, m source.Message, r source.Result) error { + var entry goredis.XMessage + if !m.As(&entry) { + return fmt.Errorf("redis: settle: message is not a redis stream entry") + } + defer s.settled() + + switch r.Action { + case source.ActionAck: + return s.ack(ctx, entry.ID) + case source.ActionNak: + // Leave the entry in the PEL; a future pending scan (NakRedeliver) claims + // and redelivers it once it has been idle long enough. The requeue delay + // raises that idle floor when larger than the configured minimum. + return nil + case source.ActionTerm: + if err := s.deadLetter(ctx, m, entry, r); err != nil { + return err + } + return s.ack(ctx, entry.ID) + case source.ActionInProgress: + return nil // Redis has no per-message deadline to extend + case source.ActionManual: + return nil + default: + return fmt.Errorf("redis: settle: unknown action %v", r.Action) + } +} + +// settled decrements the in-flight counter under the lock. It is deferred from +// Settle so a graceful Close can observe when the last delivered entry is done. +func (s *subscription) settled() { + s.mu.Lock() + if s.inflight > 0 { + s.inflight-- + } + s.mu.Unlock() +} + +// ack removes an entry from the group's pending list via XACK. +func (s *subscription) ack(ctx context.Context, id string) error { + if err := s.client.XAck(ctx, s.stream, s.group, id).Err(); err != nil { + return fmt.Errorf("redis: ack %s: %w", id, err) + } + return nil +} + +// deadLetter appends the rejected entry's fields plus dead-letter metadata +// (original ID, stream, classification, error) to the configured dead-letter +// stream. When no dead-letter stream is configured it is a no-op, so the caller +// then acks and drops the entry. +func (s *subscription) deadLetter(ctx context.Context, m source.Message, entry goredis.XMessage, r source.Result) error { + if s.dlqStream == "" { + return nil + } + values := make(map[string]any, len(entry.Values)+4) + for k, v := range entry.Values { + values[k] = v + } + values["crucible-dlq-original-id"] = entry.ID + values["crucible-dlq-stream"] = m.Subject() + values["crucible-dlq-class"] = r.Class.String() + if r.Err != nil { + values["crucible-dlq-error"] = r.Err.Error() + } + if err := s.client.XAdd(ctx, &goredis.XAddArgs{ + Stream: s.dlqStream, + ID: "*", + Values: values, + }).Err(); err != nil { + return fmt.Errorf("redis: dead-letter %s to %q: %w", entry.ID, s.dlqStream, err) + } + return nil +} + +// NakRedeliver scans the group's pending entries and claims those idle longer +// than minIdle to this consumer, returning them for redelivery. It is how a +// naked entry (left in the PEL by [source.ActionNak]) comes back: a consumer +// periodically drains the backlog with XPENDING + XCLAIM. The claimed entries +// are buffered so the next [subscription.Next] yields them. It reports how many +// entries were reclaimed. +// +// Drive it on a timer or between read cycles; the Hopper does not call it +// automatically because redelivery cadence is a deployment choice. +func (s *subscription) NakRedeliver(ctx context.Context, minIdle time.Duration) (int, error) { + if minIdle <= 0 { + minIdle = s.minIdle + } + pend, err := s.client.XPendingExt(ctx, &goredis.XPendingExtArgs{ + Stream: s.stream, + Group: s.group, + Idle: minIdle, + Start: "-", + End: "+", + Count: s.count, + }).Result() + if err != nil { + if errors.Is(err, goredis.Nil) { + return 0, nil + } + return 0, fmt.Errorf("redis: pending scan: %w", err) + } + if len(pend) == 0 { + return 0, nil + } + ids := make([]string, 0, len(pend)) + for _, p := range pend { + ids = append(ids, p.ID) + } + claimed, err := s.client.XClaim(ctx, &goredis.XClaimArgs{ + Stream: s.stream, + Group: s.group, + Consumer: s.consumer, + MinIdle: minIdle, + Messages: ids, + }).Result() + if err != nil { + return 0, fmt.Errorf("redis: claim pending: %w", err) + } + s.mu.Lock() + for _, e := range claimed { + s.pending = append(s.pending, newMessage(s.stream, e)) + } + s.mu.Unlock() + return len(claimed), nil +} + +// Close begins a graceful drain: Next stops reading new entries, and once the +// buffer is empty Next returns [source.ErrDrained]. Entries already delivered by +// Next are left for the caller to settle. Close is idempotent. +func (s *subscription) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + s.closed = true + return nil +} + +// Durable reports the consumer group name, satisfying [source.SharedDurable]. +func (s *subscription) Durable() string { return s.group } + +// SeekToTime repositions delivery to the first entry at or after t. Redis entry +// IDs embed a millisecond timestamp, so the seek translates t into the ID +// "-0" and resumes the backlog from there; it takes effect on the next +// Next. +func (s *subscription) SeekToTime(ctx context.Context, t time.Time) error { + id := fmt.Sprintf("%d-0", t.UnixMilli()) + return s.reseek(ctx, id) +} + +// SeekToCursor repositions delivery to resume from just after c, a previously +// observed [source.Cursor] carrying an entry ID. It drains the backlog from that +// ID forward using XRANGE. +func (s *subscription) SeekToCursor(ctx context.Context, c source.Cursor) error { + cur, ok := c.(idCursor) + if !ok { + return fmt.Errorf("redis: seek: cursor is not a redis cursor") + } + return s.reseek(ctx, string(cur)) +} + +// SeekToStart repositions delivery to the earliest retained entry. +func (s *subscription) SeekToStart(ctx context.Context) error { + return s.reseek(ctx, "0") +} + +// SeekToEnd repositions delivery to the tail, so only entries produced after the +// seek are delivered. It clears any buffered backlog and reads new entries. +func (s *subscription) SeekToEnd(_ context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return errSeekClosed + } + s.pending = nil + s.readFrom = ">" + return nil +} + +// reseek loads the backlog at or after fromID with XRANGE and buffers it so the +// next Next yields the replayed entries before resuming live delivery. The +// entries are claimed to this consumer (they enter the PEL) so they settle +// through the normal ack path. The caller passes a context for the range read. +func (s *subscription) reseek(ctx context.Context, fromID string) error { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return errSeekClosed + } + s.mu.Unlock() + + entries, err := s.client.XRange(ctx, s.stream, fromID, "+").Result() + if err != nil { + return fmt.Errorf("redis: seek range from %s: %w", fromID, err) + } + msgs := make([]source.Message, 0, len(entries)) + for _, e := range entries { + msgs = append(msgs, newMessage(s.stream, e)) + } + s.mu.Lock() + s.pending = msgs + s.readFrom = ">" + s.mu.Unlock() + return nil +} + +// Lag reports the consumer group's backlog behind the stream tail, satisfying +// [source.LagReporter]. It reads the group's lag from XINFO GROUPS when Redis +// reports it; otherwise it falls back to the stream length from XLEN. +func (s *subscription) Lag(ctx context.Context) (int64, error) { + groups, err := s.client.XInfoGroups(ctx, s.stream).Result() + if err == nil { + for _, g := range groups { + if g.Name == s.group { + if g.Lag > 0 { + return g.Lag, nil + } + break + } + } + } + n, err := s.client.XLen(ctx, s.stream).Result() + if err != nil { + return 0, fmt.Errorf("redis: lag: %w", err) + } + return n, nil +} + +// compile-time capability assertions: this adapter honestly advertises exactly +// the Redis-shaped capabilities and no others (no ConsumerGroups, no +// Transactional). +var ( + _ source.Subscription = (*subscription)(nil) + _ source.SharedDurable = (*subscription)(nil) + _ source.Seekable = (*subscription)(nil) + _ source.LagReporter = (*subscription)(nil) +) diff --git a/source/redis/redis_test.go b/source/redis/redis_test.go new file mode 100644 index 0000000..05629eb --- /dev/null +++ b/source/redis/redis_test.go @@ -0,0 +1,821 @@ +// SPDX-License-Identifier: Apache-2.0 + +package redis + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + goredis "github.com/redis/go-redis/v9" + + "github.com/stablekernel/crucible/source" +) + +// --- fake: a Redis Streams seam with no live server ------------------------- + +// fakeClient implements the Client seam over in-memory state, recording the +// calls a test asserts against. It is safe for concurrent Settle. +type fakeClient struct { + mu sync.Mutex + + // read returns successive batches of entries; one batch per XReadGroup call, + // then goredis.Nil to model an idle block window. + readBatches [][]goredis.XMessage + readIdx int + readErr error + + groupCreated bool + createErr error + + acked []string + ackErr error + added []*goredis.XAddArgs + addErr error + rangeOut []goredis.XMessage + rangeErr error + xlen int64 + xlenErr error + groups []goredis.XInfoGroup + groupsErr error + + pendingOut []goredis.XPendingExt + pendingErr error + claimOut []goredis.XMessage + claimErr error + + lastReadArgs *goredis.XReadGroupArgs + lastClaimArgs *goredis.XClaimArgs +} + +func (f *fakeClient) XGroupCreateMkStream(ctx context.Context, _, _, _ string) *goredis.StatusCmd { + cmd := goredis.NewStatusCmd(ctx) + f.mu.Lock() + defer f.mu.Unlock() + f.groupCreated = true + if f.createErr != nil { + cmd.SetErr(f.createErr) + } + return cmd +} + +func (f *fakeClient) XReadGroup(ctx context.Context, a *goredis.XReadGroupArgs) *goredis.XStreamSliceCmd { + cmd := goredis.NewXStreamSliceCmd(ctx) + f.mu.Lock() + defer f.mu.Unlock() + f.lastReadArgs = a + if f.readErr != nil { + cmd.SetErr(f.readErr) + return cmd + } + if f.readIdx >= len(f.readBatches) { + cmd.SetErr(goredis.Nil) + return cmd + } + batch := f.readBatches[f.readIdx] + f.readIdx++ + stream := a.Streams[0] + cmd.SetVal([]goredis.XStream{{Stream: stream, Messages: batch}}) + return cmd +} + +func (f *fakeClient) XAck(ctx context.Context, _, _ string, ids ...string) *goredis.IntCmd { + cmd := goredis.NewIntCmd(ctx) + f.mu.Lock() + defer f.mu.Unlock() + if f.ackErr != nil { + cmd.SetErr(f.ackErr) + return cmd + } + f.acked = append(f.acked, ids...) + cmd.SetVal(int64(len(ids))) + return cmd +} + +func (f *fakeClient) XPendingExt(ctx context.Context, a *goredis.XPendingExtArgs) *goredis.XPendingExtCmd { + cmd := goredis.NewXPendingExtCmd(ctx) + f.mu.Lock() + defer f.mu.Unlock() + if f.pendingErr != nil { + cmd.SetErr(f.pendingErr) + return cmd + } + cmd.SetVal(f.pendingOut) + return cmd +} + +func (f *fakeClient) XClaim(ctx context.Context, a *goredis.XClaimArgs) *goredis.XMessageSliceCmd { + cmd := goredis.NewXMessageSliceCmd(ctx) + f.mu.Lock() + defer f.mu.Unlock() + f.lastClaimArgs = a + if f.claimErr != nil { + cmd.SetErr(f.claimErr) + return cmd + } + cmd.SetVal(f.claimOut) + return cmd +} + +func (f *fakeClient) XAdd(ctx context.Context, a *goredis.XAddArgs) *goredis.StringCmd { + cmd := goredis.NewStringCmd(ctx) + f.mu.Lock() + defer f.mu.Unlock() + if f.addErr != nil { + cmd.SetErr(f.addErr) + return cmd + } + f.added = append(f.added, a) + cmd.SetVal("dlq-1") + return cmd +} + +func (f *fakeClient) XRange(ctx context.Context, _, _, _ string) *goredis.XMessageSliceCmd { + cmd := goredis.NewXMessageSliceCmd(ctx) + f.mu.Lock() + defer f.mu.Unlock() + if f.rangeErr != nil { + cmd.SetErr(f.rangeErr) + return cmd + } + cmd.SetVal(f.rangeOut) + return cmd +} + +func (f *fakeClient) XLen(ctx context.Context, _ string) *goredis.IntCmd { + cmd := goredis.NewIntCmd(ctx) + f.mu.Lock() + defer f.mu.Unlock() + if f.xlenErr != nil { + cmd.SetErr(f.xlenErr) + return cmd + } + cmd.SetVal(f.xlen) + return cmd +} + +func (f *fakeClient) XInfoGroups(ctx context.Context, _ string) *goredis.XInfoGroupsCmd { + cmd := goredis.NewXInfoGroupsCmd(ctx, "") + f.mu.Lock() + defer f.mu.Unlock() + if f.groupsErr != nil { + cmd.SetErr(f.groupsErr) + return cmd + } + cmd.SetVal(f.groups) + return cmd +} + +// entry is a terse XMessage constructor for tests. +func entry(id string, fields map[string]any) goredis.XMessage { + return goredis.XMessage{ID: id, Values: fields} +} + +// newSub builds an Inlet over a fake client and opens a subscription on +// "orders". WithBlock is short so an idle read returns promptly in tests. +func newSub(t *testing.T, c Client, opts ...Option) (*subscription, *Inlet) { + t.Helper() + base := []Option{ + WithClient(c), WithGroup("workers"), WithConsumer("w-1"), + WithBlock(10 * time.Millisecond), + } + in, err := New(append(base, opts...)...) + if err != nil { + t.Fatalf("New() error = %v", err) + } + s, err := in.Subscribe(context.Background(), source.SubscribeConfig{Topics: []string{"orders"}}) + if err != nil { + t.Fatalf("Subscribe() error = %v", err) + } + t.Cleanup(func() { _ = s.Close() }) + return s.(*subscription), in +} + +// --- New / option validation ------------------------------------------------ + +func TestNew_Validation(t *testing.T) { + t.Parallel() + tests := []struct { + name string + opts []Option + wantErr string + }{ + {name: "missing group", opts: []Option{WithAddr("x:6379"), WithConsumer("c")}, wantErr: "WithGroup is required"}, + {name: "missing consumer", opts: []Option{WithAddr("x:6379"), WithGroup("g")}, wantErr: "WithConsumer is required"}, + { + name: "addr and client mutually exclusive", + opts: []Option{WithGroup("g"), WithConsumer("c"), WithAddr("x:6379"), WithClient(&fakeClient{})}, + wantErr: "mutually exclusive", + }, + {name: "no transport", opts: []Option{WithGroup("g"), WithConsumer("c")}, wantErr: "WithAddr or WithClient is required"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + _, err := New(tt.opts...) + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("New() error = %v, want containing %q", err, tt.wantErr) + } + }) + } +} + +func TestNew_DefaultsApplied(t *testing.T) { + t.Parallel() + in, err := New(WithClient(&fakeClient{}), WithGroup("g"), WithConsumer("c")) + if err != nil { + t.Fatalf("New() error = %v", err) + } + if in.cfg.block != defaultBlock || in.cfg.count != defaultCount || in.cfg.minIdle != defaultMinIdle { + t.Errorf("defaults not applied: %+v", in.cfg) + } +} + +func TestSubscribe_RequiresExactlyOneStream(t *testing.T) { + t.Parallel() + in, _ := New(WithClient(&fakeClient{}), WithGroup("g"), WithConsumer("c")) + for _, topics := range [][]string{nil, {"a", "b"}} { + if _, err := in.Subscribe(context.Background(), source.SubscribeConfig{Topics: topics}); err == nil { + t.Errorf("Subscribe(%v) = nil, want error", topics) + } + } +} + +// --- Next: message mapping -------------------------------------------------- + +func TestNext_MapsEntry(t *testing.T) { + t.Parallel() + c := &fakeClient{readBatches: [][]goredis.XMessage{{ + entry("1526919030474-55", map[string]any{ + ValueField: "payload", KeyHeader: "cust-7", "trace": "abc", + }), + }}} + sub, _ := newSub(t, c) + + m, err := sub.Next(context.Background()) + if err != nil { + t.Fatalf("Next() error = %v", err) + } + if string(m.Key()) != "cust-7" { + t.Errorf("Key() = %q, want cust-7 (from KeyHeader)", m.Key()) + } + if string(m.Value()) != "payload" { + t.Errorf("Value() = %q, want payload", m.Value()) + } + if m.Subject() != "orders" { + t.Errorf("Subject() = %q, want orders", m.Subject()) + } + if m.PartitionKey() != "" { + t.Errorf("PartitionKey() = %q, want empty (Redis has no partitions)", m.PartitionKey()) + } + if got := m.Cursor().String(); got != "1526919030474-55" { + t.Errorf("Cursor() = %q, want the entry ID", got) + } + if v, ok := m.Headers().Get("trace"); !ok || v != "abc" { + t.Errorf("Headers().Get(trace) = %q,%v want abc,true", v, ok) + } + var xm goredis.XMessage + if !m.As(&xm) || xm.ID != "1526919030474-55" { + t.Errorf("As(*goredis.XMessage) failed, got %+v", xm) + } +} + +func TestNext_KeyFallsBackToStream(t *testing.T) { + t.Parallel() + c := &fakeClient{readBatches: [][]goredis.XMessage{{ + entry("1-0", map[string]any{ValueField: "x"}), + }}} + sub, _ := newSub(t, c) + m, err := sub.Next(context.Background()) + if err != nil { + t.Fatalf("Next() error = %v", err) + } + if string(m.Key()) != "orders" { + t.Errorf("Key() = %q, want orders (stream fallback)", m.Key()) + } +} + +func TestNext_EnsuresGroupOnce(t *testing.T) { + t.Parallel() + c := &fakeClient{readBatches: [][]goredis.XMessage{ + {entry("1-0", map[string]any{ValueField: "a"})}, + {entry("2-0", map[string]any{ValueField: "b"})}, + }} + sub, _ := newSub(t, c) + for i := 0; i < 2; i++ { + if _, err := sub.Next(context.Background()); err != nil { + t.Fatalf("Next() error = %v", err) + } + } + if !c.groupCreated { + t.Error("group was not created") + } +} + +func TestNext_GroupCreateBusyGroupIsIdempotent(t *testing.T) { + t.Parallel() + c := &fakeClient{ + createErr: errors.New("BUSYGROUP Consumer Group name already exists"), + readBatches: [][]goredis.XMessage{{entry("1-0", map[string]any{ValueField: "a"})}}, + } + sub, _ := newSub(t, c) + if _, err := sub.Next(context.Background()); err != nil { + t.Fatalf("Next() error = %v, want BUSYGROUP tolerated", err) + } +} + +func TestNext_GroupCreateRealErrorPropagates(t *testing.T) { + t.Parallel() + c := &fakeClient{createErr: errors.New("NOPERM")} + sub, _ := newSub(t, c) + if _, err := sub.Next(context.Background()); err == nil || !strings.Contains(err.Error(), "create group") { + t.Fatalf("Next() = %v, want create-group error", err) + } +} + +func TestNext_ContextCanceled(t *testing.T) { + t.Parallel() + sub, _ := newSub(t, &fakeClient{}) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := sub.Next(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Next(canceled) = %v, want context.Canceled", err) + } +} + +func TestNext_ReadErrorPropagates(t *testing.T) { + t.Parallel() + c := &fakeClient{readErr: errors.New("connection refused")} + sub, _ := newSub(t, c) + if _, err := sub.Next(context.Background()); err == nil || !strings.Contains(err.Error(), "read group") { + t.Fatalf("Next() = %v, want read-group error", err) + } +} + +func TestNext_DrainedAfterClose(t *testing.T) { + t.Parallel() + sub, _ := newSub(t, &fakeClient{}) + if err := sub.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + if _, err := sub.Next(context.Background()); !errors.Is(err, source.ErrDrained) { + t.Fatalf("Next() after Close = %v, want ErrDrained", err) + } +} + +func TestNext_DrainsBufferAfterClose(t *testing.T) { + t.Parallel() + c := &fakeClient{readBatches: [][]goredis.XMessage{{ + entry("1-0", map[string]any{ValueField: "a"}), + entry("2-0", map[string]any{ValueField: "b"}), + }}} + sub, _ := newSub(t, c) + // Buffer the batch with one Next. + if _, err := sub.Next(context.Background()); err != nil { + t.Fatalf("first Next() error = %v", err) + } + _ = sub.Close() + // The second buffered entry still drains. + if _, err := sub.Next(context.Background()); err != nil { + t.Fatalf("buffered Next() after Close = %v, want the buffered entry", err) + } + if _, err := sub.Next(context.Background()); !errors.Is(err, source.ErrDrained) { + t.Fatalf("Next() = %v, want ErrDrained once buffer empty", err) + } +} + +func TestSubscribe_GroupOverride(t *testing.T) { + t.Parallel() + c := &fakeClient{readBatches: [][]goredis.XMessage{{entry("1-0", map[string]any{ValueField: "a"})}}} + in, _ := New(WithClient(c), WithGroup("default"), WithConsumer("w-1"), WithBlock(10*time.Millisecond)) + s, _ := in.Subscribe(context.Background(), source.SubscribeConfig{Topics: []string{"orders"}, Group: "override"}) + t.Cleanup(func() { _ = s.Close() }) + if _, err := s.Next(context.Background()); err != nil { + t.Fatalf("Next() error = %v", err) + } + if c.lastReadArgs.Group != "override" { + t.Errorf("read group = %q, want override", c.lastReadArgs.Group) + } + if sd, ok := s.(source.SharedDurable); !ok || sd.Durable() != "override" { + t.Errorf("Durable() = %v,%v want override", sd, ok) + } +} + +// --- Settle: action -> Redis vocabulary ------------------------------------- + +func TestSettle_ActionMapping(t *testing.T) { + t.Parallel() + tests := []struct { + name string + result source.Result + check func(t *testing.T, c *fakeClient) + }{ + { + name: "ack calls XAck", + result: source.Ack(), + check: func(t *testing.T, c *fakeClient) { + if len(c.acked) != 1 || c.acked[0] != "1-0" { + t.Errorf("acked = %v, want [1-0]", c.acked) + } + }, + }, + { + name: "nak leaves the entry pending (no ack, no dlq)", + result: source.Nak(errors.New("transient")), + check: func(t *testing.T, c *fakeClient) { + if len(c.acked) != 0 || len(c.added) != 0 { + t.Errorf("nak settled: acked=%v added=%v, want neither", c.acked, c.added) + } + }, + }, + { + name: "term dead-letters then acks", + result: source.Term(errors.New("bad data")), + check: func(t *testing.T, c *fakeClient) { + if len(c.added) != 1 { + t.Fatalf("added = %v, want one DLQ entry", c.added) + } + v := c.added[0].Values.(map[string]any) + if v["crucible-dlq-error"] != "bad data" || v["crucible-dlq-class"] != "poison" { + t.Errorf("DLQ metadata = %v", v) + } + if v["crucible-dlq-original-id"] != "1-0" { + t.Errorf("DLQ original-id = %v, want 1-0", v["crucible-dlq-original-id"]) + } + if len(c.acked) != 1 { + t.Errorf("acked = %v, want the original acked after DLQ", c.acked) + } + }, + }, + { + name: "reject (invalid for state) dead-letters with its class", + result: source.Reject(errors.New("wrong state")), + check: func(t *testing.T, c *fakeClient) { + if len(c.added) != 1 { + t.Fatalf("added = %v, want one DLQ entry", c.added) + } + v := c.added[0].Values.(map[string]any) + if v["crucible-dlq-class"] != "invalid_for_state" { + t.Errorf("DLQ class = %v, want invalid_for_state", v["crucible-dlq-class"]) + } + }, + }, + { + name: "in progress is a no-op", + result: source.InProgress(), + check: func(t *testing.T, c *fakeClient) { + if len(c.acked) != 0 || len(c.added) != 0 { + t.Errorf("in-progress settled: acked=%v added=%v", c.acked, c.added) + } + }, + }, + { + name: "manual is a no-op", + result: source.Manual(), + check: func(t *testing.T, c *fakeClient) { + if len(c.acked) != 0 || len(c.added) != 0 { + t.Errorf("manual settled: acked=%v added=%v", c.acked, c.added) + } + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + c := &fakeClient{readBatches: [][]goredis.XMessage{{entry("1-0", map[string]any{ValueField: "a"})}}} + sub, _ := newSub(t, c, WithDLQStream("orders.dlq")) + m, err := sub.Next(context.Background()) + if err != nil { + t.Fatalf("Next() error = %v", err) + } + if err := sub.Settle(context.Background(), m, tt.result); err != nil { + t.Fatalf("Settle() error = %v", err) + } + tt.check(t, c) + }) + } +} + +func TestSettle_TermWithoutDLQJustAcks(t *testing.T) { + t.Parallel() + c := &fakeClient{readBatches: [][]goredis.XMessage{{entry("1-0", map[string]any{ValueField: "a"})}}} + sub, _ := newSub(t, c) // no WithDLQStream + m, _ := sub.Next(context.Background()) + if err := sub.Settle(context.Background(), m, source.Term(errors.New("x"))); err != nil { + t.Fatalf("Settle() error = %v", err) + } + if len(c.added) != 0 { + t.Errorf("added = %v, want no DLQ entry without WithDLQStream", c.added) + } + if len(c.acked) != 1 { + t.Errorf("acked = %v, want the entry acked-and-dropped", c.acked) + } +} + +func TestSettle_AckErrorWrapped(t *testing.T) { + t.Parallel() + boom := errors.New("unreachable") + c := &fakeClient{ + readBatches: [][]goredis.XMessage{{entry("1-0", map[string]any{ValueField: "a"})}}, + ackErr: boom, + } + sub, _ := newSub(t, c) + m, _ := sub.Next(context.Background()) + if err := sub.Settle(context.Background(), m, source.Ack()); !errors.Is(err, boom) { + t.Fatalf("Settle() = %v, want wrapped ack error", err) + } +} + +func TestSettle_DLQErrorWrapped(t *testing.T) { + t.Parallel() + boom := errors.New("dlq down") + c := &fakeClient{ + readBatches: [][]goredis.XMessage{{entry("1-0", map[string]any{ValueField: "a"})}}, + addErr: boom, + } + sub, _ := newSub(t, c, WithDLQStream("dlq")) + m, _ := sub.Next(context.Background()) + if err := sub.Settle(context.Background(), m, source.Term(errors.New("x"))); !errors.Is(err, boom) { + t.Fatalf("Settle() = %v, want wrapped dlq error", err) + } +} + +func TestSettle_UnknownAction(t *testing.T) { + t.Parallel() + c := &fakeClient{readBatches: [][]goredis.XMessage{{entry("1-0", map[string]any{ValueField: "a"})}}} + sub, _ := newSub(t, c) + m, _ := sub.Next(context.Background()) + if err := sub.Settle(context.Background(), m, source.Result{Action: source.Action(99)}); err == nil { + t.Fatal("Settle(unknown action) = nil, want error") + } +} + +func TestSettle_NonRedisMessage(t *testing.T) { + t.Parallel() + sub, _ := newSub(t, &fakeClient{}) + if err := sub.Settle(context.Background(), foreignMsg{}, source.Ack()); err == nil { + t.Fatal("Settle(foreign message) = nil, want error") + } +} + +// --- NakRedeliver: pending scan + claim ------------------------------------- + +func TestNakRedeliver_ClaimsAndBuffersPending(t *testing.T) { + t.Parallel() + c := &fakeClient{ + pendingOut: []goredis.XPendingExt{{ID: "1-0", Consumer: "other", RetryCount: 1}}, + claimOut: []goredis.XMessage{entry("1-0", map[string]any{ValueField: "redelivered"})}, + } + sub, _ := newSub(t, c) + n, err := sub.NakRedeliver(context.Background(), 0) + if err != nil { + t.Fatalf("NakRedeliver() error = %v", err) + } + if n != 1 { + t.Fatalf("NakRedeliver() = %d, want 1", n) + } + if c.lastClaimArgs == nil || c.lastClaimArgs.MinIdle != defaultMinIdle { + t.Errorf("claim MinIdle = %v, want default", c.lastClaimArgs) + } + // The claimed entry is buffered, so the next Next yields it. + m, err := sub.Next(context.Background()) + if err != nil { + t.Fatalf("Next() after redeliver error = %v", err) + } + if string(m.Value()) != "redelivered" { + t.Errorf("Value() = %q, want redelivered", m.Value()) + } +} + +func TestNakRedeliver_NoPendingIsNoOp(t *testing.T) { + t.Parallel() + sub, _ := newSub(t, &fakeClient{}) + n, err := sub.NakRedeliver(context.Background(), time.Second) + if err != nil || n != 0 { + t.Fatalf("NakRedeliver() = %d,%v want 0,nil", n, err) + } +} + +func TestNakRedeliver_PendingErrorPropagates(t *testing.T) { + t.Parallel() + c := &fakeClient{pendingErr: errors.New("boom")} + sub, _ := newSub(t, c) + if _, err := sub.NakRedeliver(context.Background(), time.Second); err == nil { + t.Fatal("NakRedeliver() = nil, want pending-scan error") + } +} + +func TestNakRedeliver_ClaimErrorPropagates(t *testing.T) { + t.Parallel() + c := &fakeClient{ + pendingOut: []goredis.XPendingExt{{ID: "1-0"}}, + claimErr: errors.New("boom"), + } + sub, _ := newSub(t, c) + if _, err := sub.NakRedeliver(context.Background(), time.Second); err == nil { + t.Fatal("NakRedeliver() = nil, want claim error") + } +} + +// --- capabilities: seek (replay by entry ID) -------------------------------- + +func TestSeek_BuffersBacklog(t *testing.T) { + t.Parallel() + seekTime := time.UnixMilli(1526919030000).UTC() + tests := []struct { + name string + seek func(s source.Seekable) error + }{ + {name: "to start", seek: func(s source.Seekable) error { return s.SeekToStart(context.Background()) }}, + {name: "to time", seek: func(s source.Seekable) error { return s.SeekToTime(context.Background(), seekTime) }}, + {name: "to cursor", seek: func(s source.Seekable) error { return s.SeekToCursor(context.Background(), idCursor("5-0")) }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + c := &fakeClient{rangeOut: []goredis.XMessage{ + entry("5-0", map[string]any{ValueField: "replayed"}), + }} + sub, _ := newSub(t, c) + sk := source.Seekable(sub) + if err := tt.seek(sk); err != nil { + t.Fatalf("seek error = %v", err) + } + m, err := sub.Next(context.Background()) + if err != nil { + t.Fatalf("Next() after seek error = %v", err) + } + if string(m.Value()) != "replayed" { + t.Errorf("Value() = %q, want replayed from backlog", m.Value()) + } + }) + } +} + +func TestSeekToEnd_SkipsBacklog(t *testing.T) { + t.Parallel() + c := &fakeClient{ + rangeOut: []goredis.XMessage{entry("1-0", map[string]any{ValueField: "old"})}, + readBatches: [][]goredis.XMessage{{entry("9-0", map[string]any{ValueField: "new"})}}, + } + sub, _ := newSub(t, c) + // Pre-load a backlog via a SeekToStart, then SeekToEnd must clear it. + if err := sub.SeekToStart(context.Background()); err != nil { + t.Fatalf("SeekToStart() error = %v", err) + } + if err := sub.SeekToEnd(context.Background()); err != nil { + t.Fatalf("SeekToEnd() error = %v", err) + } + m, err := sub.Next(context.Background()) + if err != nil { + t.Fatalf("Next() error = %v", err) + } + if string(m.Value()) != "new" { + t.Errorf("Value() = %q, want new (backlog skipped)", m.Value()) + } +} + +func TestSeek_OnClosedSubscription(t *testing.T) { + t.Parallel() + sub, _ := newSub(t, &fakeClient{}) + _ = sub.Close() + if err := sub.SeekToStart(context.Background()); !errors.Is(err, errSeekClosed) { + t.Fatalf("SeekToStart(closed) = %v, want errSeekClosed", err) + } + if err := sub.SeekToEnd(context.Background()); !errors.Is(err, errSeekClosed) { + t.Fatalf("SeekToEnd(closed) = %v, want errSeekClosed", err) + } +} + +func TestSeekToCursor_RejectsForeignCursor(t *testing.T) { + t.Parallel() + sub, _ := newSub(t, &fakeClient{}) + if err := sub.SeekToCursor(context.Background(), foreignCursor{}); err == nil { + t.Fatal("SeekToCursor(foreign) = nil, want error") + } +} + +func TestSeek_RangeErrorPropagates(t *testing.T) { + t.Parallel() + c := &fakeClient{rangeErr: errors.New("boom")} + sub, _ := newSub(t, c) + if err := sub.SeekToStart(context.Background()); err == nil { + t.Fatal("SeekToStart() = nil, want range error") + } +} + +// --- capabilities: lag ------------------------------------------------------ + +func TestLag_PrefersGroupLag(t *testing.T) { + t.Parallel() + c := &fakeClient{ + groups: []goredis.XInfoGroup{{Name: "workers", Lag: 12}}, + xlen: 999, + } + sub, _ := newSub(t, c) + got, err := sub.Lag(context.Background()) + if err != nil { + t.Fatalf("Lag() error = %v", err) + } + if got != 12 { + t.Errorf("Lag() = %d, want 12 (group lag)", got) + } +} + +func TestLag_FallsBackToXLen(t *testing.T) { + t.Parallel() + c := &fakeClient{groupsErr: errors.New("no groups info"), xlen: 7} + sub, _ := newSub(t, c) + got, err := sub.Lag(context.Background()) + if err != nil { + t.Fatalf("Lag() error = %v", err) + } + if got != 7 { + t.Errorf("Lag() = %d, want 7 (XLen fallback)", got) + } +} + +func TestLag_XLenErrorPropagates(t *testing.T) { + t.Parallel() + c := &fakeClient{groupsErr: errors.New("x"), xlenErr: errors.New("boom")} + sub, _ := newSub(t, c) + if _, err := sub.Lag(context.Background()); err == nil { + t.Fatal("Lag() = nil, want XLen error") + } +} + +// --- Inlet As / Close ------------------------------------------------------- + +func TestInlet_As_ReturnsClientSeam(t *testing.T) { + t.Parallel() + fc := &fakeClient{} + in, err := New(WithClient(fc), WithGroup("g"), WithConsumer("c")) + if err != nil { + t.Fatalf("New() error = %v", err) + } + var got Client + if !in.As(&got) || got != fc { + t.Fatalf("As(*Client) did not return the client") + } + var wrong *int + if in.As(&wrong) { + t.Fatal("As(**int) = true, want false") + } + // A fake is not a *redis.Client, so the concrete escape hatch misses. + var rc *goredis.Client + if in.As(&rc) { + t.Fatal("As(**redis.Client) = true for a fake client, want false") + } +} + +func TestInlet_Close_LeavesCallerClientOpen(t *testing.T) { + t.Parallel() + fc := &closableClient{fakeClient: &fakeClient{}} + in, _ := New(WithClient(fc), WithGroup("g"), WithConsumer("c")) + if err := in.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + if fc.closed { + t.Error("Close() shut down a caller-owned client; want it left open") + } +} + +func TestClose_Idempotent(t *testing.T) { + t.Parallel() + sub, _ := newSub(t, &fakeClient{}) + if err := sub.Close(); err != nil { + t.Fatalf("first Close() error = %v", err) + } + if err := sub.Close(); err != nil { + t.Fatalf("second Close() error = %v", err) + } +} + +// --- helpers / foreign types ------------------------------------------------ + +// closableClient adds a recording Close to the fake so the ownership test can +// assert a caller-owned client is never shut down. +type closableClient struct { + *fakeClient + closed bool +} + +func (c *closableClient) Close() error { c.closed = true; return nil } + +type foreignCursor struct{} + +func (foreignCursor) String() string { return "foreign" } + +// foreignMsg is a source.Message not backed by a redis entry. +type foreignMsg struct{} + +func (foreignMsg) Key() []byte { return nil } +func (foreignMsg) Value() []byte { return nil } +func (foreignMsg) Headers() source.Headers { return nil } +func (foreignMsg) Subject() string { return "" } +func (foreignMsg) PartitionKey() string { return "" } +func (foreignMsg) Cursor() source.Cursor { return idCursor("") } +func (foreignMsg) As(any) bool { return false }