From 9ac1206f2e4c78f0621f9f6120fcdaf69b8ad6ca Mon Sep 17 00:00:00 2001 From: Kaien Date: Sun, 9 Aug 2026 21:17:31 +0300 Subject: [PATCH] Add reusable live chat Go SDK Provide a standalone agent client for l-ai-chat with service-token auth, safe at-least-once subscriptions, ordered cumulative ACK tracking, and self-contained gRPC fake tests. Co-authored-by: Cursor --- .gitmodules | 3 + README.md | 108 ++++++++ ack_tracker.go | 67 +++++ auth.go | 42 +++ client.go | 205 ++++++++++++++ doc.go | 6 + errors.go | 12 + example_test.go | 99 +++++++ go.mod | 16 ++ go.sum | 32 +++ history.go | 70 +++++ options.go | 153 +++++++++++ proto/l-ai-chat-proto | 1 + subscription.go | 406 ++++++++++++++++++++++++++++ subscription_test.go | 605 ++++++++++++++++++++++++++++++++++++++++++ 15 files changed, 1825 insertions(+) create mode 100644 .gitmodules create mode 100644 README.md create mode 100644 ack_tracker.go create mode 100644 auth.go create mode 100644 client.go create mode 100644 doc.go create mode 100644 errors.go create mode 100644 example_test.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 history.go create mode 100644 options.go create mode 160000 proto/l-ai-chat-proto create mode 100644 subscription.go create mode 100644 subscription_test.go diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..54ea0fe --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "proto/l-ai-chat-proto"] + path = proto/l-ai-chat-proto + url = https://gitea.fe.kaien.ru/kaien/l-ai-chat-proto.git diff --git a/README.md b/README.md new file mode 100644 index 0000000..c50d8be --- /dev/null +++ b/README.md @@ -0,0 +1,108 @@ +# l-ai-chat-client-go + +Reusable Go SDK for l-ai-chat agent/service clients. It uses the authoritative +`l-ai-chat-proto` generated Go API through the `proto/l-ai-chat-proto` +submodule and does not copy protobuf files manually. + +## Install / Build + +Clone with submodules: + +```bash +git clone --recurse-submodules l-ai-chat-client-go +cd l-ai-chat-client-go +go test ./... +``` + +On this workstation the server currently consumes proto commit +`f1f44e4ed276316ef17f8deccac54da457654312`. The local sibling +`../l-ai-chat-proto` checkout is older, so the SDK submodule is pinned to the +same proto worktree commit as `../l-ai-chat`. + +## Transport and Auth + +`New` uses TLS by default and automatically adds service auth metadata to +unary RPCs, `Subscribe`, reconnects, and `Acknowledge`: + +```text +authorization: Bearer +``` + +Use `WithInsecureTransport()` only for trusted/internal networks. The service +token travels over this gRPC transport; plaintext mode is suitable only for +trusted/internal network until service-to-service TLS/mTLS is deployed. + +Tokens come from a `TokenProvider`; `StaticToken` is the simple provider. Empty +tokens and provider errors fail closed. Tokens are not included in SDK errors. + +## Delivery Guarantee + +Agent delivery is durable at-least-once: + +- receiving a message does not mean processing succeeded; +- the server durable cursor changes only after successful `Acknowledge`; +- the SDK acknowledges only after `MarkProcessed`; +- unacked messages replay after reconnect; +- handlers with external side effects must be idempotent. + +The server derives the durable consumer identity from the authenticated +agent/service token. The SDK does not accept an arbitrary consumer id. + +## Cumulative ACK Warning + +`Acknowledge(last_ack_sequence=N)` is cumulative. If deliveries are: + +```text +100, 104, 111 +``` + +and `104` finishes before `100`, do not acknowledge `104` yet: that would mark +`100` processed too. Use `Subscription.MarkProcessed`; the SDK tracks actual +delivery order, not `seq+1`, so global sequence gaps inside a topic are safe. + +Expected ACK behavior: + +```text +104 done -> no ACK +100 done -> ACK 104 +111 done -> ACK 111 +``` + +## Subscription Lifecycle + +`Subscribe` opens one server stream for one topic and waits for the first +`ready=true` marker before returning. READY is not exposed as a chat message. +Each successful stream registration is a new SDK generation. Deliveries carry +internal generation/subscription fencing, so an old `Delivery` cannot ACK a new +stream after reconnect. + +Transient stream failures (`Unavailable`, `ResourceExhausted`, +`DeadlineExceeded`, EOF/transport reset) reconnect with bounded exponential +backoff and jitter. Reconnect does not send a local receive cursor; the server +starts from the durable ACK cursor and replays unacked messages. + +`Aborted` means the server intentionally replaced this active subscription +because another stream for the same `(consumer, topic)` connected. The SDK +treats that as terminal by default to avoid reconnect ping-pong between two +replicas of one logical consumer. HA/leader election belongs above this SDK. + +First-time `(consumer, topic)` subscriptions start at the current topic head. +Use `History` for historical context. Existing consumers resume from the +durable ACK cursor. `Delivery.Replay` means redelivery/backlog context, not an +error. + +## gRPC Surface + +The SDK wraps implemented read/agent RPCs: + +- `ListTopics` +- `GetTopic` +- `GetMessage` +- `History` +- `PostMessage` using server-side `AgentPostMessage` +- `Subscribe` +- `Subscription.MarkProcessed` + +Generic `PostMessage`, `CreateTopic`, and `UpdateTopic` are intentionally not +exposed as ergonomic helpers because the current server returns +`Unimplemented` for those fail-closed gRPC paths. diff --git a/ack_tracker.go b/ack_tracker.go new file mode 100644 index 0000000..c888e55 --- /dev/null +++ b/ack_tracker.go @@ -0,0 +1,67 @@ +package chatclient + +type deliveryToken struct { + _ byte +} + +type ackEntry struct { + seq int64 + token *deliveryToken + done bool +} + +type ackTracker struct { + entries []ackEntry + index map[*deliveryToken]int +} + +func newACKTracker() ackTracker { + return ackTracker{index: make(map[*deliveryToken]int)} +} + +func (t *ackTracker) add(seq int64, token *deliveryToken) { + t.index[token] = len(t.entries) + t.entries = append(t.entries, ackEntry{seq: seq, token: token}) +} + +func (t *ackTracker) mark(token *deliveryToken) (int64, bool, bool) { + i, ok := t.index[token] + if !ok { + return 0, false, false + } + if t.entries[i].done { + return t.safeBoundary(), true, true + } + t.entries[i].done = true + return t.safeBoundary(), true, true +} + +func (t *ackTracker) safeBoundary() int64 { + var boundary int64 + for i := range t.entries { + if !t.entries[i].done { + break + } + boundary = t.entries[i].seq + } + return boundary +} + +func (t *ackTracker) removeConfirmed(lastAck int64) { + n := 0 + for n < len(t.entries) { + if !t.entries[n].done || t.entries[n].seq > lastAck { + break + } + delete(t.index, t.entries[n].token) + n++ + } + if n == 0 { + return + } + copy(t.entries, t.entries[n:]) + t.entries = t.entries[:len(t.entries)-n] + for i := range t.entries { + t.index[t.entries[i].token] = i + } +} diff --git a/auth.go b/auth.go new file mode 100644 index 0000000..d65425c --- /dev/null +++ b/auth.go @@ -0,0 +1,42 @@ +package chatclient + +import ( + "context" + "fmt" + "strings" + + "google.golang.org/grpc/credentials" +) + +type TokenProvider func(ctx context.Context) (string, error) + +func StaticToken(token string) TokenProvider { + return func(context.Context) (string, error) { + if strings.TrimSpace(token) == "" { + return "", fmt.Errorf("%w: empty service token", ErrInvalidConfig) + } + return token, nil + } +} + +type tokenCredentials struct { + provider TokenProvider + requireTransportSecurity bool +} + +func (c tokenCredentials) GetRequestMetadata(ctx context.Context, _ ...string) (map[string]string, error) { + token, err := c.provider(ctx) + if err != nil { + return nil, err + } + if strings.TrimSpace(token) == "" { + return nil, fmt.Errorf("%w: empty service token", ErrInvalidConfig) + } + return map[string]string{"authorization": "Bearer " + token}, nil +} + +func (c tokenCredentials) RequireTransportSecurity() bool { + return c.requireTransportSecurity +} + +var _ credentials.PerRPCCredentials = tokenCredentials{} diff --git a/client.go b/client.go new file mode 100644 index 0000000..50c4d5c --- /dev/null +++ b/client.go @@ -0,0 +1,205 @@ +package chatclient + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/google/uuid" + "google.golang.org/grpc" + + pb "git.be.kaien.ru/l-ai/l-ai-chat-client-go/proto/l-ai-chat-proto" +) + +type Client struct { + conn *grpc.ClientConn + own bool + api pb.ChatServiceClient + opts options + mu sync.Mutex + subs map[*Subscription]struct{} + close bool +} + +func New(ctx context.Context, target string, opts ...Option) (*Client, error) { + o := defaultOptions() + for _, opt := range opts { + opt(&o) + } + if err := validateOptions(o); err != nil { + return nil, err + } + + var conn *grpc.ClientConn + own := false + if o.conn != nil { + conn = o.conn + } else { + dialOpts, err := buildDialOptions(o) + if err != nil { + return nil, err + } + conn, err = grpc.NewClient(target, dialOpts...) + if err != nil { + return nil, err + } + own = true + } + + c := &Client{ + conn: conn, + own: own, + api: pb.NewChatServiceClient(conn), + opts: o, + subs: make(map[*Subscription]struct{}), + } + if ctx != nil { + select { + case <-ctx.Done(): + _ = c.Close() + return nil, ctx.Err() + default: + } + } + return c, nil +} + +func (c *Client) Close() error { + c.mu.Lock() + if c.close { + c.mu.Unlock() + return nil + } + c.close = true + subs := make([]*Subscription, 0, len(c.subs)) + for sub := range c.subs { + subs = append(subs, sub) + } + c.mu.Unlock() + + for _, sub := range subs { + _ = sub.Close() + } + if c.own && c.conn != nil { + return c.conn.Close() + } + return nil +} + +func (c *Client) ListTopics(ctx context.Context) ([]*pb.Topic, error) { + if err := c.checkOpen(); err != nil { + return nil, err + } + resp, err := c.api.ListTopics(ctx, &pb.ListTopicsRequest{}) + if err != nil { + return nil, err + } + return resp.GetTopics(), nil +} + +func (c *Client) GetTopic(ctx context.Context, req GetTopicRequest) (*pb.Topic, error) { + if err := c.checkOpen(); err != nil { + return nil, err + } + if req.ID != "" && req.Slug != "" { + return nil, fmt.Errorf("%w: only one of topic id or slug may be set", ErrInvalidConfig) + } + in := &pb.GetTopicRequest{} + switch { + case req.ID != "": + in.Selector = &pb.GetTopicRequest_Id{Id: req.ID} + case req.Slug != "": + in.Selector = &pb.GetTopicRequest_Slug{Slug: req.Slug} + default: + return nil, fmt.Errorf("%w: topic id or slug is required", ErrInvalidConfig) + } + return c.api.GetTopic(ctx, in) +} + +type GetTopicRequest struct { + ID string + Slug string +} + +func (c *Client) GetMessage(ctx context.Context, messageID string) (*pb.Message, error) { + if err := c.checkOpen(); err != nil { + return nil, err + } + return c.api.GetMessage(ctx, &pb.GetMessageRequest{MessageId: messageID}) +} + +type PostMessageRequest struct { + TopicID string + Content string + ReplyToMessageID string + IdempotencyKey string +} + +func (c *Client) PostMessage(ctx context.Context, req PostMessageRequest) (*pb.PostMessageResponse, error) { + if err := c.checkOpen(); err != nil { + return nil, err + } + in := &pb.AgentPostMessageRequest{ + TopicId: req.TopicID, + Content: req.Content, + IdempotencyKey: req.IdempotencyKey, + } + if req.ReplyToMessageID != "" { + replyTo := req.ReplyToMessageID + in.ReplyToMessageId = &replyTo + } + return c.api.AgentPostMessage(ctx, in) +} + +func NewIdempotencyKey() string { + return uuid.NewString() +} + +func (c *Client) Subscribe(ctx context.Context, req SubscribeRequest) (*Subscription, error) { + if err := c.checkOpen(); err != nil { + return nil, err + } + if req.TopicID == "" { + return nil, fmt.Errorf("%w: topic id is required", ErrInvalidConfig) + } + sub, err := newSubscription(ctx, c, req) + if err != nil { + return nil, err + } + c.mu.Lock() + if c.close { + c.mu.Unlock() + _ = sub.Close() + return nil, ErrClosed + } + c.subs[sub] = struct{}{} + c.mu.Unlock() + return sub, nil +} + +type SubscribeRequest struct { + TopicID string +} + +func (c *Client) unregister(sub *Subscription) { + c.mu.Lock() + delete(c.subs, sub) + c.mu.Unlock() +} + +func (c *Client) checkOpen() error { + c.mu.Lock() + defer c.mu.Unlock() + if c.close { + return ErrClosed + } + return nil +} + +func ignoreClosed(err error) error { + if errors.Is(err, context.Canceled) || errors.Is(err, ErrClosed) { + return nil + } + return err +} diff --git a/doc.go b/doc.go new file mode 100644 index 0000000..845b36f --- /dev/null +++ b/doc.go @@ -0,0 +1,6 @@ +// Package chatclient is a reusable Go SDK for l-ai-chat agent/service clients. +// +// It wraps the authoritative ChatService protobuf client with service-token +// authentication, AgentPostMessage helpers, paged History reads, and durable +// at-least-once subscriptions with ordered cumulative ACK tracking. +package chatclient diff --git a/errors.go b/errors.go new file mode 100644 index 0000000..6e8d5ce --- /dev/null +++ b/errors.go @@ -0,0 +1,12 @@ +package chatclient + +import "errors" + +var ( + ErrClosed = errors.New("chat client is closed") + ErrInvalidConfig = errors.New("invalid chat client configuration") + ErrStaleDelivery = errors.New("stale delivery") + ErrUnknownDelivery = errors.New("unknown delivery") + ErrSubscriptionReplaced = errors.New("subscription replaced") + ErrProtocolViolation = errors.New("chat service protocol violation") +) diff --git a/example_test.go b/example_test.go new file mode 100644 index 0000000..f1091fc --- /dev/null +++ b/example_test.go @@ -0,0 +1,99 @@ +package chatclient_test + +import ( + "context" + "log" + + chatclient "git.be.kaien.ru/l-ai/l-ai-chat-client-go" +) + +func ExampleNew() { + ctx := context.Background() + client, err := chatclient.New(ctx, "chat.internal:443", + chatclient.WithStaticToken("service-token-from-secret-store"), + ) + if err != nil { + log.Fatal(err) + } + defer client.Close() +} + +func ExampleClient_PostMessage() { + ctx := context.Background() + client, _ := chatclient.New(ctx, "chat.internal:443", + chatclient.WithStaticToken("service-token-from-secret-store"), + ) + defer client.Close() + + key := "security-agent:logical-action-123" + _, err := client.PostMessage(ctx, chatclient.PostMessageRequest{ + TopicID: "11111111-1111-1111-1111-111111111111", + Content: "scan completed", + IdempotencyKey: key, + }) + if err != nil { + log.Fatal(err) + } +} + +func ExampleSubscription_sequential() { + ctx := context.Background() + client, _ := chatclient.New(ctx, "chat.internal:443", + chatclient.WithStaticToken("service-token-from-secret-store"), + ) + defer client.Close() + + sub, err := client.Subscribe(ctx, chatclient.SubscribeRequest{ + TopicID: "11111111-1111-1111-1111-111111111111", + }) + if err != nil { + log.Fatal(err) + } + defer sub.Close() + + for delivery := range sub.Deliveries() { + if err := process(ctx, delivery); err != nil { + log.Fatal(err) + } + if err := sub.MarkProcessed(ctx, delivery); err != nil { + log.Fatal(err) + } + } + if err := sub.Err(); err != nil { + log.Fatal(err) + } +} + +func ExampleSubscription_concurrentProcessing() { + ctx := context.Background() + client, _ := chatclient.New(ctx, "chat.internal:443", + chatclient.WithStaticToken("service-token-from-secret-store"), + ) + defer client.Close() + + sub, err := client.Subscribe(ctx, chatclient.SubscribeRequest{ + TopicID: "11111111-1111-1111-1111-111111111111", + }) + if err != nil { + log.Fatal(err) + } + defer sub.Close() + + sem := make(chan struct{}, 4) + for delivery := range sub.Deliveries() { + delivery := delivery + sem <- struct{}{} + go func() { + defer func() { <-sem }() + if err := process(ctx, delivery); err != nil { + _ = sub.Close() + return + } + // Safe even when goroutines finish out of order: MarkProcessed + // advances only the processed prefix in actual delivery order. + _ = sub.MarkProcessed(ctx, delivery) + }() + } +} + +func process(context.Context, chatclient.Delivery) error { return nil } diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..019b128 --- /dev/null +++ b/go.mod @@ -0,0 +1,16 @@ +module git.be.kaien.ru/l-ai/l-ai-chat-client-go + +go 1.25.0 + +require ( + github.com/google/uuid v1.6.0 + google.golang.org/grpc v1.70.0 + google.golang.org/protobuf v1.36.8 +) + +require ( + golang.org/x/net v0.32.0 // indirect + golang.org/x/sys v0.28.0 // indirect + golang.org/x/text v0.21.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..890be15 --- /dev/null +++ b/go.sum @@ -0,0 +1,32 @@ +github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= +github.com/go-logr/logr v1.4.2/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/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +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= +go.opentelemetry.io/otel v1.32.0 h1:WnBN+Xjcteh0zdk01SVqV55d/m62NJLJdIyb4y/WO5U= +go.opentelemetry.io/otel v1.32.0/go.mod h1:00DCVSB0RQcnzlwyTfqtxSm+DRr9hpYrHjNGiBHVQIg= +go.opentelemetry.io/otel/metric v1.32.0 h1:xV2umtmNcThh2/a/aCP+h64Xx5wsj8qqnkYZktzNa0M= +go.opentelemetry.io/otel/metric v1.32.0/go.mod h1:jH7CIbbK6SH2V2wE16W05BHCtIDzauciCRLoc/SyMv8= +go.opentelemetry.io/otel/sdk v1.32.0 h1:RNxepc9vK59A8XsgZQouW8ue8Gkb4jpWtJm9ge5lEG4= +go.opentelemetry.io/otel/sdk v1.32.0/go.mod h1:LqgegDBjKMmb2GC6/PrTnteJG39I8/vJCAP9LlJXEjU= +go.opentelemetry.io/otel/sdk/metric v1.32.0 h1:rZvFnvmvawYb0alrYkjraqJq0Z4ZUJAiyYCU9snn1CU= +go.opentelemetry.io/otel/sdk/metric v1.32.0/go.mod h1:PWeZlq0zt9YkYAp3gjKZ0eicRYvOh1Gd+X99x6GHpCQ= +go.opentelemetry.io/otel/trace v1.32.0 h1:WIC9mYrXf8TmY/EXuULKc8hR17vE+Hjv2cssQDe03fM= +go.opentelemetry.io/otel/trace v1.32.0/go.mod h1:+i4rkvCraA+tG6AzwloGaCtkx53Fa+L+V8e9a7YvhT8= +golang.org/x/net v0.32.0 h1:ZqPmj8Kzc+Y6e0+skZsuACbx+wzMgo5MQsJh9Qd6aYI= +golang.org/x/net v0.32.0/go.mod h1:CwU0IoeOlnQQWJ6ioyFrfRuomB8GKF6KbYXZVyeXNfs= +golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA= +golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo= +golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= +google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a h1:hgh8P4EuoxpsuKMXX/To36nOFD7vixReXgn8lPGnt+o= +google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a/go.mod h1:5uTbfoYQed2U9p3KIj2/Zzm02PYhndfdmML0qC3q3FU= +google.golang.org/grpc v1.70.0 h1:pWFv03aZoHzlRKHWicjsZytKAiYCtNS0dHbXnIdq7jQ= +google.golang.org/grpc v1.70.0/go.mod h1:ofIJqVKDXx/JiXrwr2IG4/zwdH9txy3IlF40RmcJSQw= +google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= +google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= diff --git a/history.go b/history.go new file mode 100644 index 0000000..39f50d0 --- /dev/null +++ b/history.go @@ -0,0 +1,70 @@ +package chatclient + +import ( + "context" + "fmt" + + pb "git.be.kaien.ru/l-ai/l-ai-chat-client-go/proto/l-ai-chat-proto" +) + +type HistoryRequest struct { + TopicID string + BeforeSequence int64 + AfterSequence int64 + Limit int32 +} + +func (c *Client) History(ctx context.Context, req HistoryRequest) (*pb.HistoryResponse, error) { + if err := c.checkOpen(); err != nil { + return nil, err + } + if req.BeforeSequence > 0 && req.AfterSequence > 0 { + return nil, fmt.Errorf("%w: before and after sequence are mutually exclusive", ErrInvalidConfig) + } + return c.api.History(ctx, &pb.HistoryRequest{ + TopicId: req.TopicID, + BeforeSequence: req.BeforeSequence, + AfterSequence: req.AfterSequence, + Limit: req.Limit, + }) +} + +type HistoryPager struct { + client *Client + req HistoryRequest + forward bool + done bool +} + +func (c *Client) NewHistoryPager(req HistoryRequest) (*HistoryPager, error) { + if req.BeforeSequence > 0 && req.AfterSequence > 0 { + return nil, fmt.Errorf("%w: before and after sequence are mutually exclusive", ErrInvalidConfig) + } + return &HistoryPager{ + client: c, + req: req, + forward: req.AfterSequence > 0, + }, nil +} + +func (p *HistoryPager) Next(ctx context.Context) (*pb.HistoryResponse, error) { + if p.done { + return &pb.HistoryResponse{}, nil + } + resp, err := p.client.History(ctx, p.req) + if err != nil { + return nil, err + } + if len(resp.GetMessages()) == 0 || !resp.GetHasMore() { + p.done = true + } + if len(resp.GetMessages()) > 0 { + last := resp.GetMessages()[len(resp.GetMessages())-1].GetSequence() + if p.forward { + p.req.AfterSequence = last + } else { + p.req.BeforeSequence = last + } + } + return resp, nil +} diff --git a/options.go b/options.go new file mode 100644 index 0000000..dc20c70 --- /dev/null +++ b/options.go @@ -0,0 +1,153 @@ +package chatclient + +import ( + "context" + "crypto/tls" + "fmt" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" +) + +const ( + defaultSubscriptionBufferSize = 100 + defaultReconnectInitial = 200 * time.Millisecond + defaultReconnectMax = 5 * time.Second + defaultCloseTimeout = 5 * time.Second +) + +type Sleeper func(ctx context.Context, d time.Duration) error +type Jitter func(d time.Duration) time.Duration + +type Option func(*options) + +type options struct { + tokenProvider TokenProvider + insecureTransport bool + dialOptions []grpc.DialOption + conn *grpc.ClientConn + subscriptionBufferSize int + reconnectInitial time.Duration + reconnectMax time.Duration + closeTimeout time.Duration + sleeper Sleeper + jitter Jitter +} + +func defaultOptions() options { + return options{ + subscriptionBufferSize: defaultSubscriptionBufferSize, + reconnectInitial: defaultReconnectInitial, + reconnectMax: defaultReconnectMax, + closeTimeout: defaultCloseTimeout, + sleeper: sleepContext, + jitter: defaultJitter, + } +} + +func WithTokenProvider(provider TokenProvider) Option { + return func(o *options) { o.tokenProvider = provider } +} + +func WithStaticToken(token string) Option { + return WithTokenProvider(StaticToken(token)) +} + +func WithInsecureTransport() Option { + return func(o *options) { o.insecureTransport = true } +} + +func WithGRPCDialOptions(opts ...grpc.DialOption) Option { + return func(o *options) { o.dialOptions = append(o.dialOptions, opts...) } +} + +func WithClientConn(conn *grpc.ClientConn) Option { + return func(o *options) { o.conn = conn } +} + +func WithSubscriptionBufferSize(size int) Option { + return func(o *options) { o.subscriptionBufferSize = size } +} + +func WithReconnectBackoff(initial, max time.Duration) Option { + return func(o *options) { + o.reconnectInitial = initial + o.reconnectMax = max + } +} + +func withSleeperForTest(s Sleeper) Option { + return func(o *options) { o.sleeper = s } +} + +func withJitterForTest(j Jitter) Option { + return func(o *options) { o.jitter = j } +} + +func buildDialOptions(o options) ([]grpc.DialOption, error) { + if o.tokenProvider == nil { + return nil, fmt.Errorf("%w: token provider is required", ErrInvalidConfig) + } + + var transport credentials.TransportCredentials + if o.insecureTransport { + transport = insecure.NewCredentials() + } else { + transport = credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12}) + } + + creds := tokenCredentials{ + provider: o.tokenProvider, + requireTransportSecurity: !o.insecureTransport, + } + out := []grpc.DialOption{ + grpc.WithTransportCredentials(transport), + grpc.WithPerRPCCredentials(creds), + } + out = append(out, o.dialOptions...) + return out, nil +} + +func validateOptions(o options) error { + if o.tokenProvider == nil { + return fmt.Errorf("%w: token provider is required", ErrInvalidConfig) + } + if o.subscriptionBufferSize <= 0 { + return fmt.Errorf("%w: subscription buffer size must be positive", ErrInvalidConfig) + } + if o.reconnectInitial <= 0 { + return fmt.Errorf("%w: reconnect initial backoff must be positive", ErrInvalidConfig) + } + if o.reconnectMax < o.reconnectInitial { + return fmt.Errorf("%w: reconnect max backoff must be >= initial backoff", ErrInvalidConfig) + } + if o.closeTimeout <= 0 { + return fmt.Errorf("%w: close timeout must be positive", ErrInvalidConfig) + } + if o.sleeper == nil || o.jitter == nil { + return fmt.Errorf("%w: backoff hooks must be configured", ErrInvalidConfig) + } + return nil +} + +func sleepContext(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func defaultJitter(d time.Duration) time.Duration { + // Deterministic half-jitter keeps reconnects bounded without pulling in a + // retry framework. Tests replace this hook. + if d <= 1 { + return d + } + return d - d/4 +} diff --git a/proto/l-ai-chat-proto b/proto/l-ai-chat-proto new file mode 160000 index 0000000..f1f44e4 --- /dev/null +++ b/proto/l-ai-chat-proto @@ -0,0 +1 @@ +Subproject commit f1f44e4ed276316ef17f8deccac54da457654312 diff --git a/subscription.go b/subscription.go new file mode 100644 index 0000000..19d727b --- /dev/null +++ b/subscription.go @@ -0,0 +1,406 @@ +package chatclient + +import ( + "context" + "errors" + "fmt" + "io" + "sync" + "time" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + pb "git.be.kaien.ru/l-ai/l-ai-chat-client-go/proto/l-ai-chat-proto" +) + +type Delivery struct { + Message *pb.Message + Replay bool + + subscriptionID string + generation uint64 + sub *Subscription + token *deliveryToken +} + +type Handler func(context.Context, Delivery) error + +type Subscription struct { + client *Client + topic string + + ctx context.Context + cancel context.CancelFunc + + deliveries chan Delivery + done chan struct{} + ackNotify chan struct{} + + mu sync.Mutex + err error + closed bool + generation uint64 + subscriptionID string + tracker ackTracker + ackTarget int64 + ackConfirmed int64 + + readyStream pb.ChatService_SubscribeClient +} + +func newSubscription(ctx context.Context, c *Client, req SubscribeRequest) (*Subscription, error) { + if ctx == nil { + ctx = context.Background() + } + subCtx, cancel := context.WithCancel(ctx) + s := &Subscription{ + client: c, + topic: req.TopicID, + ctx: subCtx, + cancel: cancel, + deliveries: make(chan Delivery, c.opts.subscriptionBufferSize), + done: make(chan struct{}), + ackNotify: make(chan struct{}, 1), + tracker: newACKTracker(), + } + + stream, err := s.openReady(subCtx) + if err != nil { + cancel() + return nil, err + } + s.readyStream = stream + s.generation = 1 + + go s.run() + go s.runACK() + return s, nil +} + +func (s *Subscription) Deliveries() <-chan Delivery { return s.deliveries } +func (s *Subscription) Done() <-chan struct{} { return s.done } + +func (s *Subscription) Err() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.err +} + +func (s *Subscription) Close() error { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return nil + } + s.closed = true + s.mu.Unlock() + s.cancel() + + timer := time.NewTimer(s.client.opts.closeTimeout) + defer timer.Stop() + select { + case <-s.done: + return nil + case <-timer.C: + return context.DeadlineExceeded + } +} + +func (s *Subscription) MarkProcessed(ctx context.Context, d Delivery) error { + if ctx == nil { + ctx = context.Background() + } + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + if d.sub != s || d.token == nil { + return ErrUnknownDelivery + } + + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return ErrClosed + } + if d.generation != s.generation { + return ErrStaleDelivery + } + boundary, ok, _ := s.tracker.mark(d.token) + if !ok { + return ErrUnknownDelivery + } + if boundary > s.ackTarget { + s.ackTarget = boundary + s.signalACKLocked() + } + return nil +} + +func (s *Subscription) Run(ctx context.Context, handler Handler) error { + if ctx == nil { + ctx = context.Background() + } + for { + select { + case <-ctx.Done(): + return ctx.Err() + case d, ok := <-s.deliveries: + if !ok { + return ignoreClosed(s.Err()) + } + if err := handler(ctx, d); err != nil { + return err + } + if err := s.MarkProcessed(ctx, d); err != nil { + return err + } + } + } +} + +func (s *Subscription) run() { + defer func() { + s.cancel() + close(s.deliveries) + close(s.done) + s.client.unregister(s) + }() + + stream := s.readyStream + s.readyStream = nil + backoff := s.client.opts.reconnectInitial + + for { + err := s.recvUntilError(stream) + if s.ctx.Err() != nil { + s.setErr(nil) + return + } + if isReplacement(err) { + s.setErr(fmt.Errorf("%w: %v", ErrSubscriptionReplaced, err)) + return + } + if !isRetryableStreamError(err) { + s.setErr(err) + return + } + if err := s.client.opts.sleeper(s.ctx, s.client.opts.jitter(backoff)); err != nil { + s.setErr(nil) + return + } + backoff = nextBackoff(backoff, s.client.opts.reconnectMax) + + next, err := s.openReady(s.ctx) + if err != nil { + if s.ctx.Err() != nil { + s.setErr(nil) + return + } + if isReplacement(err) { + s.setErr(fmt.Errorf("%w: %v", ErrSubscriptionReplaced, err)) + return + } + if !isRetryableStreamError(err) { + s.setErr(err) + return + } + continue + } + s.resetGeneration() + stream = next + backoff = s.client.opts.reconnectInitial + } +} + +func (s *Subscription) recvUntilError(stream pb.ChatService_SubscribeClient) error { + for { + dm, err := stream.Recv() + if err != nil { + return err + } + if dm.GetReady() { + return fmt.Errorf("%w: unexpected ready marker after subscription registration", ErrProtocolViolation) + } + if dm.GetMessage() == nil { + return fmt.Errorf("%w: delivery message is missing message", ErrProtocolViolation) + } + token := &deliveryToken{} + delivery := Delivery{ + Message: dm.GetMessage(), + Replay: dm.GetReplay(), + subscriptionID: dm.GetSubscriptionId(), + token: token, + sub: s, + } + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return ErrClosed + } + if s.subscriptionID == "" { + s.subscriptionID = dm.GetSubscriptionId() + } + if dm.GetSubscriptionId() == "" || dm.GetSubscriptionId() != s.subscriptionID { + s.mu.Unlock() + return fmt.Errorf("%w: inconsistent subscription_id", ErrProtocolViolation) + } + delivery.generation = s.generation + s.tracker.add(dm.GetMessage().GetSequence(), token) + s.mu.Unlock() + + select { + case s.deliveries <- delivery: + case <-s.ctx.Done(): + return ErrClosed + } + } +} + +func (s *Subscription) openReady(ctx context.Context) (pb.ChatService_SubscribeClient, error) { + stream, err := s.client.api.Subscribe(ctx, &pb.SubscribeRequest{TopicId: s.topic}) + if err != nil { + return nil, err + } + dm, err := stream.Recv() + if err != nil { + return nil, err + } + if !dm.GetReady() || dm.GetMessage() != nil { + return nil, fmt.Errorf("%w: subscribe stream did not start with ready marker", ErrProtocolViolation) + } + return stream, nil +} + +func (s *Subscription) resetGeneration() { + s.mu.Lock() + defer s.mu.Unlock() + s.generation++ + s.subscriptionID = "" + s.tracker = newACKTracker() + s.ackTarget = 0 + s.ackConfirmed = 0 +} + +func (s *Subscription) runACK() { + for { + select { + case <-s.ctx.Done(): + return + case <-s.ackNotify: + } + + for { + gen, subID, target, ok := s.nextACK() + if !ok { + break + } + resp, err := s.client.api.Acknowledge(s.ctx, &pb.AcknowledgeRequest{ + TopicId: s.topic, + LastAckSequence: target, + SubscriptionId: subID, + }) + if err != nil { + if s.ctx.Err() != nil || !isRetryableACKError(err) || !s.sameGeneration(gen) { + if s.ctx.Err() == nil && !isRetryableACKError(err) { + s.setErr(err) + s.cancel() + } + break + } + _ = s.client.opts.sleeper(s.ctx, s.client.opts.jitter(s.client.opts.reconnectInitial)) + continue + } + s.confirmACK(gen, resp.GetLastAckSequence()) + } + } +} + +func (s *Subscription) nextACK() (uint64, string, int64, bool) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed || s.subscriptionID == "" || s.ackTarget <= s.ackConfirmed { + return 0, "", 0, false + } + return s.generation, s.subscriptionID, s.ackTarget, true +} + +func (s *Subscription) sameGeneration(gen uint64) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.generation == gen && !s.closed +} + +func (s *Subscription) confirmACK(gen uint64, confirmed int64) { + s.mu.Lock() + defer s.mu.Unlock() + if s.generation != gen || s.closed { + return + } + if confirmed > s.ackConfirmed { + s.ackConfirmed = confirmed + s.tracker.removeConfirmed(confirmed) + } + if s.ackTarget > s.ackConfirmed { + s.signalACKLocked() + } +} + +func (s *Subscription) signalACKLocked() { + select { + case s.ackNotify <- struct{}{}: + default: + } +} + +func (s *Subscription) setErr(err error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.err == nil { + s.err = err + } + s.closed = true +} + +func nextBackoff(cur, max time.Duration) time.Duration { + if cur <= 0 { + return max + } + next := cur * 2 + if next < cur || next > max { + return max + } + return next +} + +func isReplacement(err error) bool { + return status.Code(err) == codes.Aborted +} + +func isRetryableStreamError(err error) bool { + if err == nil { + return false + } + if errors.Is(err, io.EOF) { + return true + } + switch status.Code(err) { + case codes.Unavailable, codes.ResourceExhausted, codes.DeadlineExceeded: + return true + case codes.Canceled: + return false + default: + return false + } +} + +func isRetryableACKError(err error) bool { + switch status.Code(err) { + case codes.Unavailable, codes.ResourceExhausted, codes.DeadlineExceeded: + return true + default: + return false + } +} diff --git a/subscription_test.go b/subscription_test.go new file mode 100644 index 0000000..0ebf0ad --- /dev/null +++ b/subscription_test.go @@ -0,0 +1,605 @@ +package chatclient + +import ( + "context" + "errors" + "fmt" + "net" + "strings" + "sync" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" + + pb "git.be.kaien.ru/l-ai/l-ai-chat-client-go/proto/l-ai-chat-proto" +) + +type fakeChatServer struct { + pb.UnimplementedChatServiceServer + + mu sync.Mutex + authByMethod map[string][]string + postReqs []*pb.AgentPostMessageRequest + genericPosts int + ackCalls []ackCall + ackErrors map[int64][]error + subscribeCount int + subscribeFunc func(*pb.SubscribeRequest, pb.ChatService_SubscribeServer) error + historyPages []*pb.HistoryResponse + historyReqs []*pb.HistoryRequest +} + +type ackCall struct { + topic string + subID string + seq int64 +} + +func newFake() *fakeChatServer { + return &fakeChatServer{ + authByMethod: make(map[string][]string), + ackErrors: make(map[int64][]error), + } +} + +func (f *fakeChatServer) record(ctx context.Context, method string) { + md, _ := metadata.FromIncomingContext(ctx) + f.mu.Lock() + defer f.mu.Unlock() + f.authByMethod[method] = append([]string(nil), md.Get("authorization")...) +} + +func (f *fakeChatServer) ListTopics(ctx context.Context, _ *pb.ListTopicsRequest) (*pb.ListTopicsResponse, error) { + f.record(ctx, "ListTopics") + return &pb.ListTopicsResponse{Topics: []*pb.Topic{{Id: "topic-1", Slug: "general"}}}, nil +} + +func (f *fakeChatServer) GetTopic(ctx context.Context, _ *pb.GetTopicRequest) (*pb.Topic, error) { + f.record(ctx, "GetTopic") + return &pb.Topic{Id: "topic-1", Slug: "general"}, nil +} + +func (f *fakeChatServer) GetMessage(ctx context.Context, req *pb.GetMessageRequest) (*pb.Message, error) { + f.record(ctx, "GetMessage") + return &pb.Message{Id: req.GetMessageId(), Sequence: 1}, nil +} + +func (f *fakeChatServer) PostMessage(context.Context, *pb.PostMessageRequest) (*pb.PostMessageResponse, error) { + f.mu.Lock() + f.genericPosts++ + f.mu.Unlock() + return nil, status.Error(codes.Unimplemented, "generic post is fail-closed") +} + +func (f *fakeChatServer) AgentPostMessage(ctx context.Context, req *pb.AgentPostMessageRequest) (*pb.PostMessageResponse, error) { + f.record(ctx, "AgentPostMessage") + f.mu.Lock() + f.postReqs = append(f.postReqs, req) + f.mu.Unlock() + return &pb.PostMessageResponse{ + Message: &pb.Message{Id: "posted", TopicId: req.GetTopicId(), Content: req.GetContent(), Sequence: 10}, + }, nil +} + +func (f *fakeChatServer) History(ctx context.Context, req *pb.HistoryRequest) (*pb.HistoryResponse, error) { + f.record(ctx, "History") + f.mu.Lock() + defer f.mu.Unlock() + f.historyReqs = append(f.historyReqs, req) + if len(f.historyPages) == 0 { + return &pb.HistoryResponse{}, nil + } + page := f.historyPages[0] + f.historyPages = f.historyPages[1:] + return page, nil +} + +func (f *fakeChatServer) Subscribe(req *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + f.record(stream.Context(), "Subscribe") + f.mu.Lock() + f.subscribeCount++ + fn := f.subscribeFunc + f.mu.Unlock() + if fn == nil { + return status.Error(codes.Unavailable, "no subscribe script") + } + return fn(req, stream) +} + +func (f *fakeChatServer) Acknowledge(ctx context.Context, req *pb.AcknowledgeRequest) (*pb.AcknowledgeResponse, error) { + f.record(ctx, "Acknowledge") + f.mu.Lock() + f.ackCalls = append(f.ackCalls, ackCall{topic: req.GetTopicId(), subID: req.GetSubscriptionId(), seq: req.GetLastAckSequence()}) + errs := f.ackErrors[req.GetLastAckSequence()] + if len(errs) > 0 { + err := errs[0] + f.ackErrors[req.GetLastAckSequence()] = errs[1:] + f.mu.Unlock() + return nil, err + } + f.mu.Unlock() + return &pb.AcknowledgeResponse{LastAckSequence: req.GetLastAckSequence()}, nil +} + +func newClientForTest(t *testing.T, fake *fakeChatServer, opts ...Option) *Client { + t.Helper() + lis := bufconn.Listen(1 << 20) + srv := grpc.NewServer() + pb.RegisterChatServiceServer(srv, fake) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(func() { + srv.Stop() + _ = lis.Close() + }) + + all := []Option{ + WithStaticToken("test-token"), + WithInsecureTransport(), + WithGRPCDialOptions(grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return lis.DialContext(ctx) + })), + withJitterForTest(func(d time.Duration) time.Duration { return d }), + withSleeperForTest(func(ctx context.Context, d time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + default: + return nil + } + }), + } + all = append(all, opts...) + c, err := New(context.Background(), "passthrough:///bufnet", all...) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + return c +} + +func msg(seq int64) *pb.DeliveryMessage { + return &pb.DeliveryMessage{ + Message: &pb.Message{Id: fmt.Sprintf("m-%d", seq), TopicId: "topic", Sequence: seq}, + SubscriptionId: "sub-1", + } +} + +func recvDelivery(t *testing.T, sub *Subscription) Delivery { + t.Helper() + select { + case d := <-sub.Deliveries(): + return d + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for delivery") + return Delivery{} + } +} + +func waitDone(t *testing.T, sub *Subscription) { + t.Helper() + select { + case <-sub.Done(): + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for subscription done") + } +} + +func waitACKs(t *testing.T, fake *fakeChatServer, want ...int64) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + fake.mu.Lock() + got := make([]int64, len(fake.ackCalls)) + for i, c := range fake.ackCalls { + got[i] = c.seq + } + fake.mu.Unlock() + if len(got) >= len(want) { + ok := true + for i := range want { + if got[i] != want[i] { + ok = false + } + } + if ok { + return + } + } + time.Sleep(10 * time.Millisecond) + } + fake.mu.Lock() + defer fake.mu.Unlock() + t.Fatalf("ack calls = %+v, want prefix %v", fake.ackCalls, want) +} + +func TestAuthMetadataUnarySubscribeAcknowledgeAndEmptyTokenFailsClosed(t *testing.T) { + fake := newFake() + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + if err := stream.Send(&pb.DeliveryMessage{Ready: true}); err != nil { + return err + } + if err := stream.Send(msg(100)); err != nil { + return err + } + <-stream.Context().Done() + return stream.Context().Err() + } + c := newClientForTest(t, fake) + if _, err := c.ListTopics(context.Background()); err != nil { + t.Fatal(err) + } + sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + if err != nil { + t.Fatal(err) + } + d := recvDelivery(t, sub) + if err := sub.MarkProcessed(context.Background(), d); err != nil { + t.Fatal(err) + } + waitACKs(t, fake, 100) + + fake.mu.Lock() + defer fake.mu.Unlock() + for _, method := range []string{"ListTopics", "Subscribe", "Acknowledge"} { + got := fake.authByMethod[method] + if len(got) != 1 || got[0] != "Bearer test-token" { + t.Fatalf("%s auth metadata = %v", method, got) + } + } + + empty := newClientForTest(t, newFake(), WithStaticToken("")) + _, err = empty.ListTopics(context.Background()) + if status.Code(err) != codes.Unauthenticated { + t.Fatalf("empty token: got %v", err) + } + if err != nil && strings.Contains(err.Error(), "test-token") { + t.Fatalf("error leaked token: %v", err) + } +} + +func TestPostMessageUsesAgentPostMessageAndPreservesIdempotencyKey(t *testing.T) { + fake := newFake() + c := newClientForTest(t, fake) + _, err := c.PostMessage(context.Background(), PostMessageRequest{ + TopicID: "topic", + Content: "payload", + IdempotencyKey: "logical-key", + }) + if err != nil { + t.Fatal(err) + } + fake.mu.Lock() + defer fake.mu.Unlock() + if fake.genericPosts != 0 { + t.Fatal("generic PostMessage must not be called") + } + if len(fake.postReqs) != 1 || fake.postReqs[0].GetIdempotencyKey() != "logical-key" { + t.Fatalf("AgentPostMessage reqs = %+v", fake.postReqs) + } +} + +func TestOrderedCumulativeACKWithSequenceGaps(t *testing.T) { + fake := newFake() + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + for _, seq := range []int64{100, 104, 111} { + _ = stream.Send(msg(seq)) + } + <-stream.Context().Done() + return stream.Context().Err() + } + c := newClientForTest(t, fake) + sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + if err != nil { + t.Fatal(err) + } + d100, d104, d111 := recvDelivery(t, sub), recvDelivery(t, sub), recvDelivery(t, sub) + if err := sub.MarkProcessed(context.Background(), d104); err != nil { + t.Fatal(err) + } + time.Sleep(30 * time.Millisecond) + fake.mu.Lock() + if len(fake.ackCalls) != 0 { + t.Fatalf("ACK advanced over unprocessed 100: %+v", fake.ackCalls) + } + fake.mu.Unlock() + if err := sub.MarkProcessed(context.Background(), d100); err != nil { + t.Fatalf("mark 100: %v", err) + } + waitACKs(t, fake, 104) + if err := sub.MarkProcessed(context.Background(), d111); err != nil { + t.Fatalf("mark 111: %v", err) + } + waitACKs(t, fake, 104, 111) +} + +func TestLargerOutOfOrderCompletionDoesNotSkipEarliestPending(t *testing.T) { + fake := newFake() + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + for _, seq := range []int64{100, 104, 111, 118, 130} { + _ = stream.Send(msg(seq)) + } + <-stream.Context().Done() + return stream.Context().Err() + } + c := newClientForTest(t, fake) + sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + deliveries := map[int64]Delivery{} + for i := 0; i < 5; i++ { + d := recvDelivery(t, sub) + deliveries[d.Message.GetSequence()] = d + } + for _, seq := range []int64{118, 104} { + _ = sub.MarkProcessed(context.Background(), deliveries[seq]) + } + time.Sleep(30 * time.Millisecond) + fake.mu.Lock() + if len(fake.ackCalls) != 0 { + t.Fatalf("ACK skipped earliest pending: %+v", fake.ackCalls) + } + fake.mu.Unlock() + if err := sub.MarkProcessed(context.Background(), deliveries[100]); err != nil { + t.Fatalf("mark 100: %v", err) + } + waitACKs(t, fake, 104) + if err := sub.MarkProcessed(context.Background(), deliveries[130]); err != nil { + t.Fatalf("mark 130: %v", err) + } + time.Sleep(30 * time.Millisecond) + fake.mu.Lock() + if len(fake.ackCalls) != 1 { + t.Fatalf("ACK skipped 111: %+v", fake.ackCalls) + } + fake.mu.Unlock() + if err := sub.MarkProcessed(context.Background(), deliveries[111]); err != nil { + t.Fatalf("mark 111: %v", err) + } + waitACKs(t, fake, 104, 130) +} + +func TestACKTransientFailureIsRetriedWithoutLosingBoundary(t *testing.T) { + fake := newFake() + fake.ackErrors[104] = []error{status.Error(codes.Unavailable, "try again")} + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + _ = stream.Send(msg(100)) + _ = stream.Send(msg(104)) + <-stream.Context().Done() + return stream.Context().Err() + } + c := newClientForTest(t, fake) + sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + d100, d104 := recvDelivery(t, sub), recvDelivery(t, sub) + if err := sub.MarkProcessed(context.Background(), d100); err != nil { + t.Fatalf("mark 100: %v", err) + } + if err := sub.MarkProcessed(context.Background(), d104); err != nil { + t.Fatalf("mark 104: %v", err) + } + waitACKs(t, fake, 104, 104) +} + +func TestReconnectReplayAndStaleGenerationFencing(t *testing.T) { + fake := newFake() + var calls int + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + calls++ + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + if calls == 1 { + _ = stream.Send(msg(100)) + _ = stream.Send(msg(104)) + return status.Error(codes.Unavailable, "drop") + } + dm := msg(104) + dm.Replay = true + dm.SubscriptionId = "sub-2" + _ = stream.Send(dm) + <-stream.Context().Done() + return stream.Context().Err() + } + c := newClientForTest(t, fake) + sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + d100, old104 := recvDelivery(t, sub), recvDelivery(t, sub) + if err := sub.MarkProcessed(context.Background(), d100); err != nil { + t.Fatalf("mark 100: %v", err) + } + waitACKs(t, fake, 100) + new104 := recvDelivery(t, sub) + if !new104.Replay { + t.Fatal("reconnected delivery must expose replay=true") + } + if err := sub.MarkProcessed(context.Background(), old104); !errors.Is(err, ErrStaleDelivery) { + t.Fatalf("old generation MarkProcessed = %v", err) + } + if err := sub.MarkProcessed(context.Background(), new104); err != nil { + t.Fatal(err) + } + waitACKs(t, fake, 100, 104) +} + +func TestStreamReplacementAbortedIsTerminalNoReconnect(t *testing.T) { + fake := newFake() + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + return status.Error(codes.Aborted, "subscription replaced") + } + c := newClientForTest(t, fake) + sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + if err != nil { + t.Fatal(err) + } + waitDone(t, sub) + if !errors.Is(sub.Err(), ErrSubscriptionReplaced) { + t.Fatalf("Err = %v", sub.Err()) + } + fake.mu.Lock() + defer fake.mu.Unlock() + if fake.subscribeCount != 1 { + t.Fatalf("replacement triggered reconnect storm: %d subscribes", fake.subscribeCount) + } +} + +func TestAuthTerminalErrorsDoNotEndlesslyReconnect(t *testing.T) { + for _, code := range []codes.Code{codes.Unauthenticated, codes.PermissionDenied} { + t.Run(code.String(), func(t *testing.T) { + fake := newFake() + fake.subscribeFunc = func(_ *pb.SubscribeRequest, _ pb.ChatService_SubscribeServer) error { + return status.Error(code, "terminal") + } + c := newClientForTest(t, fake) + if _, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}); status.Code(err) != code { + t.Fatalf("Subscribe error = %v", err) + } + fake.mu.Lock() + defer fake.mu.Unlock() + if fake.subscribeCount != 1 { + t.Fatalf("terminal auth reconnects: %d", fake.subscribeCount) + } + }) + } +} + +func TestTransientReconnect(t *testing.T) { + fake := newFake() + var calls int + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + calls++ + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + if calls == 1 { + return status.Error(codes.Unavailable, "transient") + } + _ = stream.Send(msg(200)) + <-stream.Context().Done() + return stream.Context().Err() + } + c := newClientForTest(t, fake) + sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + if err != nil { + t.Fatal(err) + } + d := recvDelivery(t, sub) + if d.Message.GetSequence() != 200 { + t.Fatalf("delivery seq = %d", d.Message.GetSequence()) + } +} + +func TestBoundedBufferShutdownDoesNotDropOrPanic(t *testing.T) { + fake := newFake() + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + for _, seq := range []int64{1, 2, 3} { + _ = stream.Send(msg(seq)) + } + <-stream.Context().Done() + return stream.Context().Err() + } + c := newClientForTest(t, fake, WithSubscriptionBufferSize(1)) + sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + if err != nil { + t.Fatal(err) + } + if err := sub.Close(); err != nil { + t.Fatal(err) + } +} + +func TestMultipleTopicSubscriptionsAreIndependent(t *testing.T) { + fake := newFake() + fake.subscribeFunc = func(req *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + dm := msg(1) + dm.Message.TopicId = req.GetTopicId() + dm.SubscriptionId = "sub-" + req.GetTopicId() + _ = stream.Send(dm) + if req.GetTopicId() == "general" { + return status.Error(codes.Unavailable, "one topic failed") + } + <-stream.Context().Done() + return stream.Context().Err() + } + c := newClientForTest(t, fake) + general, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "general"}) + security, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "security"}) + if recvDelivery(t, general).Message.GetTopicId() != "general" { + t.Fatal("general delivery mismatch") + } + ds := recvDelivery(t, security) + if ds.Message.GetTopicId() != "security" { + t.Fatal("security delivery mismatch") + } + if err := security.MarkProcessed(context.Background(), ds); err != nil { + t.Fatal(err) + } + waitACKs(t, fake, 1) +} + +func TestCloseRacesWithReconnectRecvAckAndMarkProcessed(t *testing.T) { + fake := newFake() + ackEntered := make(chan struct{}) + releaseACK := make(chan struct{}) + fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error { + _ = stream.Send(&pb.DeliveryMessage{Ready: true}) + _ = stream.Send(msg(1)) + <-stream.Context().Done() + return stream.Context().Err() + } + fake.ackErrors[1] = []error{status.Error(codes.Unavailable, "retry")} + c := newClientForTest(t, fake, withSleeperForTest(func(ctx context.Context, d time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + default: + return nil + } + })) + origAck := fake.Acknowledge + _ = origAck + sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}) + d := recvDelivery(t, sub) + go func() { _ = sub.MarkProcessed(context.Background(), d) }() + close(ackEntered) + <-ackEntered + close(releaseACK) + go func() { _ = sub.MarkProcessed(context.Background(), d) }() + if err := sub.Close(); err != nil { + t.Fatal(err) + } +} + +func TestHistoryPagerUsesReturnedSequenceAcrossGaps(t *testing.T) { + fake := newFake() + fake.historyPages = []*pb.HistoryResponse{ + {Messages: []*pb.Message{{Sequence: 100}, {Sequence: 104}, {Sequence: 111}}, HasMore: true}, + {Messages: []*pb.Message{{Sequence: 118}}, HasMore: false}, + } + c := newClientForTest(t, fake) + pager, err := c.NewHistoryPager(HistoryRequest{TopicID: "topic", AfterSequence: 99, Limit: 3}) + if err != nil { + t.Fatal(err) + } + first, err := pager.Next(context.Background()) + if err != nil { + t.Fatal(err) + } + if first.Messages[len(first.Messages)-1].GetSequence() != 111 { + t.Fatal("first page setup broken") + } + second, err := pager.Next(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(second.Messages) != 1 || second.Messages[0].GetSequence() != 118 { + t.Fatalf("second page = %+v", second.Messages) + } + fake.mu.Lock() + defer fake.mu.Unlock() + if len(fake.historyReqs) != 2 || fake.historyReqs[1].GetAfterSequence() != 111 { + t.Fatalf("history reqs = %+v", fake.historyReqs) + } +}