streaming: generation-qualified stream lifecycle and exact publication (#82)

* streaming: generation-qualified stream lifecycle and exact publication

Streams bind lazily to one Redis-owned generation with a distinct physical
key per explicit recreation. Every metadata mutation - publication, event
removal, consumer-group recovery, consumer registration, keep-alive refresh,
stale recovery, acknowledgement, and destruction - verifies the exact
generation atomically in Redis, so destroyed-generation metadata can never be
resurrected by a concurrent sink. Adds idempotent AddOnce publication and
side-effect-free Snapshot reads, Redis-owned TTL/deadline retention, and a
fenced stale-recovery lease. Exported option structs keep their v1 fields as
a stable ordered prefix; unkeyed literals are pinned internally and keyed v1
construction is pinned for external compatibility.

* streaming: address review findings

Delete the orphaned rollbackStreamRegistration helper (atomic registration
made Redis rollback unnecessary), keep the exported Stream.MaxLen immutable
after construction so concurrent readers never race with generation binding
(the canonical bound lives in the private snapshot), and pin the v1 option
fields to their leading positions with a reflect-based prefix check that
keyed literals cannot provide.

* pool: synchronize the shared test hasher in TestStaleNodeStreamCleanup

The hasher closure is shared by two nodes and invoked from concurrent
routing and rebalance goroutines; its job counter must be atomic. Latent on
main, exposed by the streaming lifecycle's timing changes under -race.

* streaming: gofmt

Author: raphael

examples/rmap/basics/main.go

@@ -14,7 +14,7 @@ func main() {
 	ctx := context.Background()
 
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 
 	// Make sure Redis is up and running and we can connect to it
 	if err := rdb.Ping(ctx).Err(); err != nil {

examples/rmap/multi-nodes/main.go

@@ -18,7 +18,7 @@ func main() {
 	ctx := context.Background()
 
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 
 	// Make sure Redis is up and running and we can connect to it
 	if err := rdb.Ping(ctx).Err(); err != nil {

examples/streaming/README.md

@@ -10,30 +10,34 @@ To run the examples, follow these steps:
 
 2. Open a terminal or command prompt.
 
+   The examples read `REDIS_ADDR` and `REDIS_PASSWORD`; from the repository
+   root, `source .env` selects the default local Redis configuration.
+
 3. Clone the `goadesign/pulse` repository by running the following command:
-   ```
+   ```bash
    git clone https://github.com/goadesign/pulse.git
    ```
 
 4. Change into the example directory (e.g. `examples/streaming/single-reader`):
-   ```
+   ```bash
    cd pulse/examples/streaming/single-reader
    ```
-   ```
 
-5. Install the required dependencies by running the following command:
-   ```
-   go get github.com/redis/go-redis/v9 goa.design/pulse/rmap
+5. Download the repository's pinned dependencies:
+   ```bash
+   go mod download
    ```
 
 6. Build the Go program by executing the following command:
-   ```
+   ```bash
    go build
    ```
 
 7. Run the program using the following command:
-   ```
+   ```bash
    ./single-reader
    ```
 
 This will execute the program and demonstrate the basic operations on streaming.
+The [`exact-publication`](exact-publication/main.go) example shows
+deadline-owned `AddOnce` retries and read-only snapshots.

examples/streaming/exact-publication/main.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	"goa.design/pulse/streaming"
+	"goa.design/pulse/streaming/options"
+)
+
+func main() {
+	ctx := context.Background()
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     os.Getenv("REDIS_ADDR"),
+		Password: os.Getenv("REDIS_PASSWORD"),
+	})
+	if err := rdb.Ping(ctx).Err(); err != nil {
+		panic(err)
+	}
+
+	deadline := time.Now().Add(time.Minute).Truncate(time.Millisecond)
+	stream, err := streaming.NewStream(
+		"exact-publication",
+		rdb,
+		options.WithStreamDeadline(deadline),
+		options.WithStreamMaxLen(1_000),
+	)
+	if err != nil {
+		panic(err)
+	}
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
+
+	first, err := stream.AddOnce(
+		ctx,
+		"facility-42:alarm-7",
+		"alarm-opened",
+		[]byte("high discharge pressure"),
+		options.WithTopic("alarms"),
+	)
+	if err != nil {
+		panic(err)
+	}
+	retry, err := stream.AddOnce(
+		ctx,
+		"facility-42:alarm-7",
+		"alarm-opened",
+		[]byte("high discharge pressure"),
+		options.WithTopic("alarms"),
+	)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Printf("first=%s retry=%s\n", first, retry)
+
+	events, err := stream.Snapshot(ctx)
+	if err != nil {
+		panic(err)
+	}
+	for _, event := range events {
+		fmt.Printf("%s %s: %s\n", event.ID(), event.EventName(), event.Payload())
+	}
+}

examples/streaming/multi-readers/main.go

@@ -13,7 +13,7 @@ import (
 
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 
 	// Make sure Redis is up and running and we can connect to it

examples/streaming/multi-sinks/main.go

@@ -14,7 +14,7 @@ import (
 // NOTE: the example below does not handle errors for brevity.
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 
 	// Make sure Redis is up and running and we can connect to it
@@ -29,7 +29,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream.Destroy(ctx)
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Write 2 events to the stream
 	id1, err := stream.Add(ctx, "event 1", []byte("payload 1"))
@@ -54,7 +58,11 @@ func main() {
 	}
 
 	// Don't forget to close the sink when done
-	defer sink1.Close(ctx)
+	defer func() {
+		if err := sink1.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Read and acknowlege event
 	ev := <-sink1.Subscribe()
@@ -70,12 +78,16 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	defer sink2.Close(ctx)
+	defer func() {
+		if err := sink2.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Read second event
 	ev = <-sink2.Subscribe()
 	fmt.Printf("sink 2, event: %s, payload: %s\n", ev.EventName, ev.Payload)
-	if sink2.Ack(ctx, ev); err != nil {
+	if err := sink2.Ack(ctx, ev); err != nil {
 		panic(err)
 	}
 }

examples/streaming/multi-streams/main.go

@@ -14,7 +14,7 @@ import (
 // Note: the example below does not handle errors for brevity.
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 
 	// Make sure Redis is up and running and we can connect to it
@@ -29,7 +29,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream1.Destroy(ctx)
+	defer func() {
+		if err := stream1.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Create sink
 	sink, err := stream1.NewSink(ctx, "multistreams-sink",
@@ -40,7 +44,11 @@ func main() {
 	}
 
 	// Don't forget to close the sink when done
-	defer sink.Close(ctx)
+	defer func() {
+		if err := sink.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Subscribe to events
 	c := sink.Subscribe()
@@ -57,7 +65,11 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	defer stream2.Destroy(ctx)
+	defer func() {
+		if err := stream2.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Add stream to sink
 	err = sink.AddStream(ctx, stream2)

examples/streaming/pub-sub/main.go

@@ -14,7 +14,7 @@ import (
 // NOTE: the example below does not handle errors for brevity.
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 	if err := rdb.Ping(ctx).Err(); err != nil {
 		panic(err)
@@ -27,7 +27,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream.Destroy(ctx)
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Add a new event to topic "my-topic"
 	id1, err := stream.Add(ctx,
@@ -56,7 +60,11 @@ func main() {
 	}
 
 	// Don't forget to close the sink when done
-	defer sink.Close(ctx)
+	defer func() {
+		if err := sink.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Read both events
 	c := sink.Subscribe()

examples/streaming/single-reader/main.go

@@ -13,7 +13,7 @@ import (
 
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 	if err := rdb.Ping(ctx).Err(); err != nil {
 		panic(err)
@@ -26,7 +26,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream.Destroy(ctx)
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Add a new event
 	id, err := stream.Add(ctx, "event", []byte("payload"))

examples/streaming/single-sink/main.go

@@ -13,7 +13,7 @@ import (
 
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 	if err := rdb.Ping(ctx).Err(); err != nil {
 		panic(err)
@@ -26,7 +26,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream.Destroy(ctx)
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Add a new event
 	id, err := stream.Add(ctx, "event", []byte("payload"))
@@ -45,7 +49,11 @@ func main() {
 	}
 
 	// Don't forget to close the sink when done
-	defer sink.Close(ctx)
+	defer func() {
+		if err := sink.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Consume event
 	ev := <-sink.Subscribe()