diff --git a/README.md b/README.md index ba523a5..aafa4ee 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,14 @@ Typed Go client for [Weir](https://github.com/batchstream/weir). ## Overview +### Packages + +| Package | Use | +| --- | --- | +| `github.com/batchstream/weir-go` | The client: `Read`, `Mutate` and `Native` for any Store | +| `github.com/batchstream/weir-go/mongodb` | Typed MongoDB collections, queries and commands | +| `github.com/batchstream/weir-go/search` | Typed Elasticsearch and OpenSearch indexes, searches and HTTP requests | + ### Terms | Term | Meaning | @@ -50,8 +58,9 @@ Typed Go client for [Weir](https://github.com/batchstream/weir). | Elasticsearch, OpenSearch | `index/s:key` | `application/json` | MongoDB keys can also use `i:` for a 64-bit integer or `oid:` for an ObjectId in -lowercase hex. `weir.EncodeSegment` encodes a path segment. The SDK does not encode -documents. Your code owns the document bytes. +lowercase hex. `weir.EncodeSegment` encodes a path segment. The `weir` package does +not encode documents; your code owns the document bytes. The `mongodb` and `search` +packages build resources and encode documents for you. ### Mutation outcomes @@ -68,7 +77,8 @@ documents. Your code owns the document bytes. finish. 2. Set a context deadline on every call. 3. Check the Go error and every result. A record can fail while the Go error is - nil. + nil. `MutationErr`, `ReadResult.Err` and `NativeResult.Err` turn a result into + an error. 4. Never automatically resend a mutation that is `MutationApplied` or `MutationUnknown`, or that has no result. It has applied or may have applied. Reconcile it first. @@ -193,7 +203,9 @@ message: hi | [Operations](docs/operations.md) | Reads, writes, visibility, results and outcomes | | [Lua](docs/lua.md) | Lua transforms | | [Native](docs/native.md) | Backend commands and HTTP helpers | -| [Streaming](docs/streaming.md) | Callbacks, streams, deadlines and resending refused records | +| [MongoDB](docs/mongodb.md) | The `mongodb` package: collections, queries and commands | +| [Search](docs/search.md) | The `search` package: indexes, searches and HTTP requests | +| [Streaming](docs/streaming.md) | Callbacks, streams, deadlines and resending records | | [Development](docs/development.md) | Tests, real backend tests and releases | | [API reference](https://pkg.go.dev/github.com/batchstream/weir-go) | Every exported type and function | @@ -202,6 +214,8 @@ message: hi | [read](examples/read/main.go) | A typed read | | [basic](examples/basic/main.go) | A mutation, then many reads in one call | | [native](examples/native/main.go) | A native command and its response | +| [mongodb](examples/mongodb/main.go) | A typed MongoDB collection | +| [search](examples/search/main.go) | A typed search index and a search | ## Development diff --git a/api.go b/api.go index 316d284..3d2f8d9 100644 --- a/api.go +++ b/api.go @@ -12,7 +12,7 @@ type Failure = pb.Failure // FailureCode identifies a declared operation failure. Only known nonzero codes // are accepted. A Client sends a record again only for FailureUnavailable on -// a record that never began. +// a record that never began, or FailureConflict on a mutation NOT_APPLIED. type FailureCode = pb.FailureCode type MutationResult = pb.MutationResult type MutationOutcome = pb.MutationOutcome diff --git a/client.go b/client.go index bc96238..97b3fa2 100644 --- a/client.go +++ b/client.go @@ -1,6 +1,7 @@ // Package weir sends typed business requests to the Stores hosted behind one -// Weir application address. A Client sends a record again only when a server -// refused it before it began; nothing that may have taken effect is resent. +// Weir application address. A Client sends a record again only when it took +// no effect: a server refused it before it began, or a mutation lost to +// concurrent writers. Nothing that may have taken effect is resent. package weir import ( diff --git a/client_stream.go b/client_stream.go index 0e36339..20d493c 100644 --- a/client_stream.go +++ b/client_stream.go @@ -33,7 +33,7 @@ type clientStreamOptions[R comparable] struct { // send returns the result of one command, sending it once unless it is // already queued. send func(context.Context, *pb.Command, *queuedRecord) (R, error) - refused func(R) bool + resend func(R) resendCause consume func(context.Context, int, R) error } @@ -229,7 +229,7 @@ func (s *clientStream[R]) run(record *streamRecord, queued *queuedRecord) { queued = nil return s.options.send(s.ctx, record.command, current) } - result, err := resendRefused(s.ctx, s.client, s.options.refused, send) + result, err := withResends(s.ctx, s.client, s.options.resend, send) if err != nil { s.fail(err) } diff --git a/client_test.go b/client_test.go index 43fa818..fccbf47 100644 --- a/client_test.go +++ b/client_test.go @@ -3,6 +3,7 @@ package weir import ( "bytes" "context" + "errors" "fmt" "io" "net" @@ -165,7 +166,7 @@ func (p *clientTestPeer) mutationRecord(request *pb.ExecuteRequest, stream grpc. } result := protocol.Mutation(MutationApplied, nil) if strings.HasPrefix(p.mode, "batch_") && request.Index == 2 { - result = protocol.Mutation(MutationNotApplied, protocol.Fail(FailureConflict, "conflict")) + result = protocol.Mutation(MutationNotApplied, protocol.Fail(FailurePreconditionFailed, "record exists")) } switch p.mode { case "invalid_outcome": @@ -361,6 +362,14 @@ func TestPutLostResponseHasNoAcknowledgementOrReplay(t *testing.T) { if status.Code(err) != codes.Unavailable || result != nil || peer.streams.Load() != 1 || peer.received.Load() != 1 { t.Fatal("lost response confirmed or replayed mutation", result, err, peer.streams.Load(), peer.received.Load()) } + if !errors.Is(err, ErrInterrupted) { + t.Fatal("lost response not marked interrupted", err) + } + other := &MutateRequest{Resource: "records/s:other", Action: MutationPut, Document: document} + results, err := client.Mutate(t.Context(), "records", WithRequests(request, other)) + if !errors.Is(err, ErrInterrupted) || status.Code(err) != codes.Unavailable || results[0] != nil { + t.Fatal("lost response in a call not marked interrupted", results, err) + } } func TestMutationRejectsInvalidOutcomeAndKeepsAppliedFailure(t *testing.T) { diff --git a/connection.go b/connection.go index 9923442..296430d 100644 --- a/connection.go +++ b/connection.go @@ -28,8 +28,8 @@ var ErrClosed = errors.New("Weir client is closed") // Client sends business requests to the Weir servers behind one application // address. Every server behind the address must host the Stores the client // uses. Reuse a Client across goroutines and close it after callers finish. -// It sends a record again only when a server refused it before it began, and -// never one that may have taken effect. +// It sends a record again only when it took no effect, as when a server +// refused it before it began, and never one that may have taken effect. type Client struct { connection *grpc.ClientConn store pb.StoreServiceClient @@ -132,7 +132,8 @@ func (c *Client) storeClient() (pb.StoreServiceClient, error) { // callError reports a call that ended with its context as the context's error, // whether it used a shared stream or a stream of its own. The server enforces -// the propagated deadline too, so its status can arrive before ctx expires. +// the propagated deadline too, so its status can arrive before ctx expires. A +// lost connection is marked ErrInterrupted and keeps its gRPC status. func callError(ctx context.Context, err error) error { if err == nil { return nil @@ -143,6 +144,9 @@ func callError(ctx context.Context, err error) error { if _, bounded := ctx.Deadline(); bounded && status.Code(err) == codes.DeadlineExceeded { return context.DeadlineExceeded } + if status.Code(err) == codes.Unavailable { + return fmt.Errorf("%w: %w", ErrInterrupted, err) + } return err } diff --git a/docs/connections.md b/docs/connections.md index 0fe49df..7d198a0 100644 --- a/docs/connections.md +++ b/docs/connections.md @@ -62,9 +62,11 @@ Reuse the client across goroutines and close it after callers finish. Calls started after `Close` return `ErrClosed`; calls still running end with a gRPC cancellation error. A call whose context ends returns the context's -error, so test it with `errors.Is` rather than its gRPC status. A Client sends a record again only when a server refused it -before it began, as described under -[Resending refused records](streaming.md#resending-refused-records). +error, so test it with `errors.Is` rather than its gRPC status. A call cut short by +a lost connection returns an error that `errors.Is(err, weir.ErrInterrupted)` +reports, with its gRPC status kept. A Client sends a record again only when it +took no effect, as described under +[Resending records](streaming.md#resending-records). ## Kubernetes DNS names diff --git a/docs/development.md b/docs/development.md index 21a13e0..4c0dbd5 100644 --- a/docs/development.md +++ b/docs/development.md @@ -2,6 +2,19 @@ How to test the SDK and publish a version. +## Packages + +| Package | Contents | +| --- | --- | +| `weir` | The client and its protocol types. It knows no backend. | +| `mongodb` | MongoDB documents and commands, built on `weir`'s public API | +| `search` | Elasticsearch and OpenSearch documents and HTTP requests, built on `weir`'s public API | +| `internal/record` | Single-record calls the two packages share | +| `internal/storetest` | A fake Weir for the two packages' tests | + +`weir` cannot depend on a backend package: both import it, and Go rejects import +cycles. + ## Module dependencies The SDK depends on the stable `github.com/batchstream/weir-protocol v0.17.0` @@ -16,6 +29,8 @@ generate schemas. mutates a document then reads multiple resources in one call, and [native](../examples/native/main.go) consumes native responses. These examples use SDK request and result types. +[mongodb](../examples/mongodb/main.go) and [search](../examples/search/main.go) use +the `mongodb` and `search` packages. ## Tests @@ -27,7 +42,7 @@ go vet ./... Default tests use owned loopback fixtures, without databases or production I/O. CI checks typed operation dispatch, acknowledgements with transport errors, malformed/empty native chunks, incremental consumption, -completion/cancellation, refusal-only resends and deadline admission. It also checks that +completion/cancellation, resends of only refused and conflicted records, and deadline admission. It also checks that the module graph does not contain the Weir server, that `go.mod` replaces no module and that the protocol is a released `vMAJOR.MINOR.PATCH` version. @@ -48,12 +63,14 @@ WEIR_MONGO_STORE=mongo \ WEIR_MONGO_RESOURCE=weir_example/records \ WEIR_SEARCH_STORE=search \ WEIR_SEARCH_RESOURCE=weir_example \ -go test -race -tags integration -run 'Test(Mongo|Search)(Lifecycle|Streaming)' -v . +go test -race -tags integration -v ./... ``` Pre-create disposable collections/indexes. Each test creates and cleans up only its generated records. Lifecycle tests cover CRUD, backend expressions and Native; streaming tests verify same-resource ordering through an individual -failure, Lua persistence, and repeated and missing reads. System integration tests, -backend fault tests and throughput benchmarks belong to the independent +failure, Lua persistence, and repeated and missing reads. The `mongodb` and +`search` packages test every method, cursors and command failures against the +same fixtures. System integration tests, backend fault tests and throughput +benchmarks belong to the independent [weir-tests](https://github.com/batchstream/weir-tests) repository. diff --git a/docs/mongodb.md b/docs/mongodb.md new file mode 100644 index 0000000..dcfbffd --- /dev/null +++ b/docs/mongodb.md @@ -0,0 +1,145 @@ +# MongoDB package + +How to read and write MongoDB documents with `github.com/batchstream/weir-go/mongodb`. + +## Setup + +A `DB` names a database behind a Weir Store whose backend is MongoDB. A +`Collection` adds a collection, the Go type of its documents and the type of +their `_id`. Both are plain values: build them once, copy and share them freely. + +```go +client, err := weir.New("127.0.0.1:7447") +if err != nil { return err } +defer client.Close() + +db := mongodb.DB{Client: client, Store: "mongo", Name: "shop"} +users := mongodb.Collection[string, User]{DB: db, Name: "users"} +``` + +| Field | Meaning | +| --- | --- | +| `DB.Store` | Name of a Weir Store whose backend is MongoDB | +| `DB.Name` | Database name | +| `Collection.Name` | Collection name. The collection must exist before its first write. | +| `Collection.QueryVisible` | Each write returns only once later queries see it | + +The `_id` type is a string, an `int`, `int32` or `int64`, or a `bson.ObjectID`, +or a type defined on a string or integer. Documents are encoded with the +MongoDB Go driver's `bson` package. + +## Records + +| Method | Effect | Failure code when it does not apply | +| --- | --- | --- | +| `Get(ctx, id)` | Returns the document, or `weir.ErrNotFound` | | +| `GetMany(ctx, ids...)` | Returns the documents in one call, in order, nil for a missing one | | +| `Create(ctx, id, doc)` | Writes a document that must not exist | `FailurePreconditionFailed`, also for a unique index | +| `Put(ctx, id, doc)` | Creates or replaces the document | | +| `Replace(ctx, id, doc)` | Replaces a document that must exist | `FailurePreconditionFailed` | +| `Delete(ctx, id)` | Deletes the document; a missing one succeeds | | +| `Update(ctx, id, update)` | Applies `$set`, `$unset` and `$inc` to a document that must exist | `FailurePreconditionFailed` | +| `Transform(ctx, id, program, input)` | Runs a Lua program atomically against the document | `FailurePreconditionFailed` for `weir.reject` | + +Writes take the `_id` separately. A document may leave out its `_id` field or +leave it zero; one it sets must equal the id. The package puts the `_id` first, +as Weir requires. + +```go +ada := &User{Name: "Ada"} +err := users.Create(ctx, "ada", ada) +if weir.CodeOf(err) == weir.FailurePreconditionFailed { + // Ada already exists. +} + +visits := bson.D{{Key: "$inc", Value: bson.D{{Key: "visits", Value: 1}}}} +err = users.Update(ctx, "ada", visits) + +user, err := users.Get(ctx, "ada") +if errors.Is(err, weir.ErrNotFound) { + // No such user. +} +``` + +`Transform` sends `input` as BSON, or no input when it is nil. The program +returns one function, called as `function(current, incoming)`. See +[Lua](lua.md) for what it may return. + +```go +const rename = `return function(current, incoming) + if current == nil then return weir.reject("no such user") end + current.name = incoming.name + return current +end` +name := bson.D{{Key: "name", Value: "Ada L."}} +err := users.Transform(ctx, "ada", rename, name) +``` + +Concurrent calls on one Client share its streams, so many goroutines writing +single records still reach Weir in batches. + +## Queries + +`Find` yields the documents that match a filter, nil for all. Further options are +fields of MongoDB's `find` command, such as `sort`, `projection`, `skip`, `limit` +and `batchSize`. It follows the cursor with `getMore`, and kills the cursor when +the loop stops early. An error ends the loop. + +```go +active := bson.D{{Key: "active", Value: true}} +newest := bson.E{Key: "sort", Value: bson.D{{Key: "_id", Value: -1}}} +for user, err := range users.Find(ctx, active, newest) { + if err != nil { return err } + fmt.Println(user.Name) +} +``` + +`Count` returns the number of matching documents. It counts them; for a cheap +estimate of a whole collection, run the `count` command with `DB.Command`. + +## Commands + +`DB.Command` runs any command in the database and returns its reply. A reply with +`ok: 0`, `writeErrors` or a `writeConcernError` comes back together with a +`*CommandError` holding its code, code name and message. `DB.Cursor` runs a +command that opens a cursor, such as `aggregate` or `listIndexes`, and yields its +documents the way `Find` does. + +```go +create := bson.D{{Key: "create", Value: "users"}} +reply, err := db.Command(ctx, create) +var failed *mongodb.CommandError +if errors.As(err, &failed) && failed.Name == "NamespaceExists" { + // The collection already exists. +} + +pipeline := bson.A{bson.D{{Key: "$group", Value: bson.D{{Key: "_id", Value: "$country"}}}}} +aggregate := bson.D{{Key: "aggregate", Value: "users"}, {Key: "pipeline", Value: pipeline}, {Key: "cursor", Value: bson.D{}}} +for document, err := range db.Cursor(ctx, aggregate) { + if err != nil { return err } + fmt.Println(document) +} +``` + +A command is sent again only when a server refused it before it began. One a +lost connection or an unavailable backend cut short reports +`weir.ErrInterrupted`, and may have run. Send it again only when that is safe, +as it is for a read. + +## Errors + +| Error | Meaning | +| --- | --- | +| `weir.ErrNotFound` | `Get` found no document | +| `*weir.Error` | A record failed. `weir.CodeOf(err)` gives its `FailureCode`, and `Outcome` tells whether a write applied. | +| `weir.ErrInterrupted` | An outage cut the request short, so it may succeed later. It may have run: never send a write again without checking. | +| `*mongodb.CommandError` | MongoDB rejected a command or one of its writes | + +A write certainly took no effect only when its `*weir.Error` has the `Outcome` +`MutationNotStarted` or `MutationNotApplied`, or, for a command, the +`Completion` `NativeNotStarted`. After any other error, a context error +included, it may have applied. + +The Client already sent again every write a server refused before it began and +every write that lost to concurrent writers of its record; see +[Streaming](streaming.md#resending-records). diff --git a/docs/native.md b/docs/native.md index 2e6b12f..7b01887 100644 --- a/docs/native.md +++ b/docs/native.md @@ -70,4 +70,7 @@ available. NativeResponseComplete means all response bytes arrived. Its status/b can report a backend error such as HTTP 404 with no Go error; interpret that data separately. A validated terminal is retained alongside a later RPC error. A Client sends a Native request again only after a refused NativeNotStarted without -a Response. +a Response. `NativeResult.Err` returns nil for a complete response and an +`*Error` otherwise; one that began and ended with UNAVAILABLE or without its end +is marked `ErrInterrupted`. A Native call cut short by a lost connection returns +an error marked `ErrInterrupted` too. diff --git a/docs/operations.md b/docs/operations.md index 5fb49dc..82f4890 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -132,12 +132,46 @@ cannot establish whether it applied. A later RPC failure does not revoke any confirmed item. Unconfirmed nil positions may have applied; reconcile them through application knowledge and never automatically replay writes. +## Result errors + +These turn results into Go errors, so a caller can return them with `err`: + +| Function | Returns nil for | Otherwise | +| --- | --- | --- | +| `MutationErr(result)` | APPLIED without a Failure | `*Error` with Outcome and Failure | +| `result.Err()` on a `ReadResult` | A document or a confirmed missing record | `*Error` with the Failure | +| `result.Err()` on a `NativeResult` | A complete response, whatever its own status | `*Error` with the Failure, if any | + +```go +results, err := client.Mutate(ctx, "search", weir.WithRequests(request)) +if err != nil { return err } +if err := weir.MutationErr(results[0]); err != nil { + if weir.CodeOf(err) == weir.FailurePreconditionFailed { + // The record already exists. + } + return err +} +``` + +`CodeOf(err)` returns the FailureCode of an `*Error` anywhere in the chain, or +zero. `errors.Is(err, ErrInterrupted)` reports a request an outage cut short, so +sending it again later may succeed: a lost connection, a mutation UNKNOWN with +UNAVAILABLE, or a Native request that began and ended with UNAVAILABLE or without +its end. It may have run, so send it again only when repeating it is safe. + +`ErrInterrupted` does not tell whether a write applied. A write certainly took no +effect only when its `*Error` has the `Outcome` `MutationNotStarted` or +`MutationNotApplied`, or the `Completion` `NativeNotStarted`. After any other +error, a context error included, it may have applied. `ErrNotFound` is what the +[mongodb](mongodb.md) and [search](search.md) packages return for a missing +record. + ## Response validation Requests and responses reject unknown protobuf fields recursively. FailureCode accepts only declared nonzero values. The SDK validates required result variants, completion evidence, document envelopes and indexes. A Failure leads to a resend only as described under -[Resending refused records](streaming.md#resending-refused-records), and never changes +[Resending records](streaming.md#resending-records), and never changes confirmed MutationApplied evidence. A malformed later response preserves previously confirmed results. diff --git a/docs/search.md b/docs/search.md new file mode 100644 index 0000000..d2fe9e0 --- /dev/null +++ b/docs/search.md @@ -0,0 +1,124 @@ +# Search package + +How to read and write Elasticsearch and OpenSearch documents with +`github.com/batchstream/weir-go/search`. + +## Setup + +A `Cluster` names a Weir Store whose backend is search. An `Index` adds an index +and the Go type of its documents. Both are plain values: build them once, copy +and share them freely. + +```go +client, err := weir.New("127.0.0.1:7447") +if err != nil { return err } +defer client.Close() + +cluster := search.Cluster{Client: client, Store: "search"} +products := search.Index[Product]{Cluster: cluster, Name: "products"} +``` + +| Field | Meaning | +| --- | --- | +| `Cluster.Store` | Name of a Weir Store whose backend is search | +| `Index.Name` | An index, or an alias of one index. It must exist before its first write. | +| `Index.QueryVisible` | Each write returns only once later searches see it | + +Document IDs are strings. Documents are encoded with `encoding/json`. + +## Records + +| Method | Effect | Failure code when it does not apply | +| --- | --- | --- | +| `Get(ctx, id)` | Returns the document, or `weir.ErrNotFound` | | +| `GetMany(ctx, ids...)` | Returns the documents in one call, in order, nil for a missing one | | +| `Create(ctx, id, doc)` | Writes a document that must not exist | `FailurePreconditionFailed` | +| `Put(ctx, id, doc)` | Creates or replaces the document | | +| `Replace(ctx, id, doc)` | Replaces a document that must exist | `FailurePreconditionFailed` | +| `Delete(ctx, id)` | Deletes the document; a missing one succeeds | | +| `Update(ctx, id, fields)` | Merges fields into a document that must exist | `FailurePreconditionFailed` | +| `Transform(ctx, id, program, input)` | Runs a Lua program atomically against the document | `FailurePreconditionFailed` for `weir.reject` | + +```go +lamp := &Product{Name: "Lamp", Price: 30} +err := products.Put(ctx, "p1", lamp) + +price := map[string]any{"price": 25} +err = products.Update(ctx, "p1", price) + +product, err := products.Get(ctx, "p1") +if errors.Is(err, weir.ErrNotFound) { + // No such product. +} +``` + +`Transform` sends `input` as JSON, or no input when it is nil. See [Lua](lua.md) +for what the program may return. + +## Searches + +`Search` sends a search request body to the index and returns its hits. Each +`Hit` has the document's `ID`, `Score`, `Sort` values and the decoded `Document`. +`Sort` keeps each value's exact JSON, so pass it back as `search_after` to read +the next page. A status of 300 or more is a `*search.StatusError`. `Search` +returns only the hits: for the total, aggregations or any other part of the +response, send the request with `Do` and decode its `Body`. + +```go +query := map[string]any{ + "query": map[string]any{"match": map[string]any{"name": "lamp"}}, + "sort": []any{map[string]any{"price": "asc"}, map[string]any{"_id": "asc"}}, + "size": 100, +} +hits, err := products.Search(ctx, query) +if err != nil { return err } +for _, hit := range hits { + fmt.Println(hit.ID, hit.Document.Name) +} +if len(hits) > 0 { + query["search_after"] = hits[len(hits)-1].Sort +} +``` + +## HTTP requests + +`Index.Do` sends any HTTP request with a path relative to the index, such as +`/_count` or `/_refresh`, and `Cluster.Do` one with a path from the cluster root, +such as `/_cluster/health`. A path may carry a query string. A `[]byte` or +`json.RawMessage` body is sent as it is, so it suits `_bulk`; any other body is +sent as JSON. The response comes back whatever its status: check `StatusCode`. + +```go +count := map[string]any{"query": map[string]any{"term": map[string]any{"name": "lamp"}}} +response, err := products.Do(ctx, http.MethodPost, "/_count", count) +if err != nil { return err } +if response.StatusCode != http.StatusOK { + return fmt.Errorf("count failed: %s", response.Body) +} +``` + +For headers of your own or a response body read in chunks, use +`weir.NewHTTPNativeRequest` and `Client.Native`; see [Native](native.md). + +A request is sent again only when a server refused it before it began. One a +lost connection or an unavailable backend cut short reports +`weir.ErrInterrupted`, and may have run. Send it again only when that is safe, +as it is for a search. + +## Errors + +| Error | Meaning | +| --- | --- | +| `weir.ErrNotFound` | `Get` found no document | +| `*weir.Error` | A record failed. `weir.CodeOf(err)` gives its `FailureCode`, and `Outcome` tells whether a write applied. | +| `weir.ErrInterrupted` | An outage cut the request short, so it may succeed later. It may have run: never send a write again without checking. | +| `*search.StatusError` | `Search` got a status of 300 or more | + +A write certainly took no effect only when its `*weir.Error` has the `Outcome` +`MutationNotStarted` or `MutationNotApplied`, or, for a request, the +`Completion` `NativeNotStarted`. After any other error, a context error +included, it may have applied. + +The Client already sent again every write a server refused before it began and +every write that lost to concurrent writers of its record; see +[Streaming](streaming.md#resending-records). diff --git a/docs/streaming.md b/docs/streaming.md index 214a96d..e04edd1 100644 --- a/docs/streaming.md +++ b/docs/streaming.md @@ -103,26 +103,33 @@ rebalance clients behind one virtual IP, and idle streams close after 10 seconds. Shared streams do not carry outgoing gRPC metadata from a call's context; a call with outgoing metadata uses a stream of its own and sends it. -## Resending refused records - -A Client sends a record again when a server refused it before it began: a -`MutationNotStarted` mutation, a read Failure, or a `NativeNotStarted` without a -Response, whose Failure code is `FailureUnavailable`. A replica draining for a -rolling restart, a replica under memory pressure and an unavailable backend -refuse this way. Such a record took no effect, so sending it again cannot apply -it twice. `FailureResourceExhausted` reports that the request exceeded a budget -of its own, such as a result over the message size limit, which a resend would -exceed again. Every other result, including UNKNOWN, NOT_APPLIED, APPLIED with a +## Resending records + +A Client sends a record again when it took no effect and may succeed later: + +| Result | Cause | +| --- | --- | +| `MutationNotStarted`, read Failure or `NativeNotStarted` without a Response, with `FailureUnavailable` | A server refused it before it began: a replica draining for a rolling restart, a replica under memory pressure or an unavailable backend | +| `MutationNotApplied` with `FailureConflict` | It lost every compare-and-swap or transaction attempt Weir made against concurrent writers of the same record | + +Such a record took no effect, so sending it again cannot apply it twice. A +resent atomic transform runs again against the record as it then is. +`FailureResourceExhausted` reports that the request exceeded a budget of its own, +such as a result over the message size limit, which a resend would exceed again. +Every other result, including UNKNOWN, other NOT_APPLIED failures, APPLIED with a Failure and nil positions, and every Go error reach the caller unchanged. A record is sent at most four times. The wait before each resend is jittered between half and all of 50 ms, 100 ms and then 200 ms. A resend is made only when the context deadline leaves that longest wait again for the send after it. A resend that still outlasts the deadline ends with the deadline error, and that -record may have taken effect. Resends across a Client stay within one per ten -first sends plus a burst of 100, so a server that refuses everything never -receives several times its normal traffic. A record still refused at the end -keeps its NOT_STARTED evidence, and later mutations of its resource in the same -call still run after it. The caller may send it again, unless such a mutation -did not end NOT_STARTED; sending it then would apply it after that mutation. Use -caller deadlines; they bound resends as well as calls. +record may have taken effect. Refusals and conflicts each have a budget: across +a Client, the resends of each cause stay within one per ten first sends plus a +burst of 100. A server that refuses everything never receives several times its +normal traffic, a heavily contended record does not multiply its writers, and +contention never uses up the resends of records a draining server refused. A +record still refused or conflicted at the end keeps +its evidence, and later mutations of its resource in the same call still run +after it. The caller may send it again, unless such a mutation did not end +NOT_STARTED or NOT_APPLIED; sending it then would apply it after that mutation. +Use caller deadlines; they bound resends as well as calls. diff --git a/errors.go b/errors.go new file mode 100644 index 0000000..e126f3d --- /dev/null +++ b/errors.go @@ -0,0 +1,129 @@ +package weir + +import ( + "errors" + "fmt" +) + +// ErrNotFound reports a record that does not exist. Read reports one as +// ReadResult.Missing; the mongodb and search packages return ErrNotFound. +var ErrNotFound = errors.New("weir: record not found") + +// ErrInterrupted marks a request an outage cut short, so that sending it again +// later may succeed: the connection to Weir broke, a mutation ended UNKNOWN +// with UNAVAILABLE, or a Native request that began ended with UNAVAILABLE or +// without its end. Whether the backend ran it is unknown, so send it again +// only when repeating it is safe, as it is for a read. A request a server +// refused before it began is not interrupted: it took no effect, and the +// Client already sent it again. Neither is a request that failed for a reason +// of its own, or whose context ended. +// +// ErrInterrupted does not tell whether a write applied. A write certainly took +// no effect only when its *Error has the Outcome MutationNotStarted or +// MutationNotApplied, or the Completion NativeNotStarted. After any other +// error, a context error included, it may have applied. +var ErrInterrupted = errors.New("weir: interrupted") + +// Error is a Failure Weir reported for one record or Native request. +type Error struct { + // Outcome is the outcome of a mutation, and zero for a read or Native + // request. + Outcome MutationOutcome + // Completion is how far a Native request got, and zero for a record or a + // Native response that stopped without its end. + Completion NativeCompletion + // Failure is nil when the request has no result or its Native response + // stopped early without one. + Failure *Failure + + reason string + interrupted bool +} + +func (e *Error) Error() string { + text := "weir" + if e.Outcome != 0 { + text += ": mutation " + e.Outcome.String() + } + if e.Completion != 0 { + text += ": native " + e.Completion.String() + } + if e.Failure != nil { + text += fmt.Sprintf(": %s: %s", e.Failure.GetCode(), e.Failure.GetMessage()) + } + if e.reason != "" { + text += ": " + e.reason + } + return text +} + +// Unwrap returns ErrInterrupted for an interrupted request. +func (e *Error) Unwrap() error { + if e.interrupted { + return ErrInterrupted + } + return nil +} + +// CodeOf returns the FailureCode of the first *Error in err's chain, or zero +// when there is none or it has no Failure. +func CodeOf(err error) FailureCode { + var failed *Error + if !errors.As(err, &failed) { + return 0 + } + return failed.Failure.GetCode() +} + +// MutationErr returns nil for an APPLIED result without a Failure, and an +// *Error otherwise. APPLIED with a Failure has applied, so never send it again. +// A nil result, a position a broken call left unconfirmed, may have applied. +func MutationErr(result *MutationResult) error { + if result == nil { + failed := &Error{Outcome: MutationUnknown, reason: "without a result", interrupted: true} + return failed + } + outcome := result.GetOutcome() + if outcome == MutationApplied && result.GetFailure() == nil { + return nil + } + interrupted := outcome == MutationUnknown && result.GetFailure().GetCode() == FailureUnavailable + failed := &Error{Outcome: outcome, Failure: result.GetFailure(), interrupted: interrupted} + return failed +} + +// Err returns nil for a read that returned a document or confirmed it +// missing, and an *Error for a read that failed or has no result. +func (r *ReadResult) Err() error { + if r == nil { + failed := &Error{reason: "read without a result", interrupted: true} + return failed + } + if r.Failure != nil { + failed := &Error{Failure: r.Failure} + return failed + } + return nil +} + +// Err returns nil for a Native request whose response arrived complete, and an +// *Error otherwise. The response's own status, such as an HTTP 404 or a +// MongoDB reply with ok: 0, is data for the caller and never an Err. +func (r *NativeResult) Err() error { + if r == nil { + failed := &Error{reason: "native request without a result", interrupted: true} + return failed + } + if r.Failure != nil { + interrupted := r.Completion != NativeNotStarted && r.Failure.GetCode() == FailureUnavailable + failed := &Error{Completion: r.Completion, Failure: r.Failure, interrupted: interrupted} + return failed + } + // A terminal that is not complete carries a Failure, so this response + // stopped without one. + if r.Completion != NativeResponseComplete { + failed := &Error{Completion: r.Completion, reason: "native response without its end", interrupted: true} + return failed + } + return nil +} diff --git a/errors_test.go b/errors_test.go new file mode 100644 index 0000000..41cd432 --- /dev/null +++ b/errors_test.go @@ -0,0 +1,112 @@ +package weir + +import ( + "context" + "errors" + "testing" + + "github.com/batchstream/weir-protocol/api/protocol" + pb "github.com/batchstream/weir-protocol/api/weir/v1" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestMutationErrAcceptsOnlyAppliedWithoutFailure(t *testing.T) { + unavailable := protocol.Fail(pb.FailureCode_UNAVAILABLE, "backend unavailable") + precondition := protocol.Fail(pb.FailureCode_PRECONDITION_FAILED, "record exists") + deadline := protocol.Fail(pb.FailureCode_DEADLINE_EXCEEDED, "deadline") + cases := []struct { + name string + result *MutationResult + ok bool + code FailureCode + interrupted bool + }{ + {name: "applied", result: protocol.Mutation(pb.MutationOutcome_APPLIED, nil), ok: true}, + {name: "applied without visibility", result: protocol.Mutation(pb.MutationOutcome_APPLIED, unavailable), code: FailureUnavailable}, + {name: "not applied", result: protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, precondition), code: FailurePreconditionFailed}, + {name: "unknown after an outage", result: protocol.Mutation(pb.MutationOutcome_UNKNOWN, unavailable), code: FailureUnavailable, interrupted: true}, + {name: "unknown at the deadline", result: protocol.Mutation(pb.MutationOutcome_UNKNOWN, deadline), code: FailureDeadlineExceeded}, + {name: "no result", interrupted: true}, + } + for _, test := range cases { + err := MutationErr(test.result) + if (err == nil) != test.ok || CodeOf(err) != test.code || errors.Is(err, ErrInterrupted) != test.interrupted { + t.Error(test.name, err) + } + var failed *Error + if err != nil && (!errors.As(err, &failed) || failed.Outcome != test.result.GetOutcome() && test.result != nil) { + t.Error(test.name, "lost its outcome", err) + } + } + text := MutationErr(protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, precondition)).Error() + if text != "weir: mutation NOT_APPLIED: PRECONDITION_FAILED: record exists" { + t.Error(text) + } +} + +func TestReadAndNativeErrs(t *testing.T) { + unavailable := protocol.Fail(pb.FailureCode_UNAVAILABLE, "backend unavailable") + invalid := protocol.Fail(pb.FailureCode_INVALID_ARGUMENT, "invalid") + head := &NativeResponse{} + found := &ReadResult{Document: &Document{}} + missing := &ReadResult{Missing: true} + failedRead := &ReadResult{Failure: unavailable} + var noRead *ReadResult + complete := &NativeResult{Response: head, Completion: NativeResponseComplete} + refused := &NativeResult{Completion: NativeNotStarted, Failure: unavailable} + lost := &NativeResult{Response: head, Completion: NativeResponseIncomplete, Failure: unavailable} + invalidNative := &NativeResult{Completion: NativeNotStarted, Failure: invalid} + endless := &NativeResult{Response: head} + var noNative *NativeResult + cases := []struct { + name string + err error + ok bool + code FailureCode + interrupted bool + }{ + {name: "found read", err: found.Err(), ok: true}, + {name: "missing read", err: missing.Err(), ok: true}, + {name: "failed read", err: failedRead.Err(), code: FailureUnavailable}, + {name: "no read result", err: noRead.Err(), interrupted: true}, + {name: "complete Native", err: complete.Err(), ok: true}, + {name: "refused Native", err: refused.Err(), code: FailureUnavailable}, + {name: "Native that lost its backend", err: lost.Err(), code: FailureUnavailable, interrupted: true}, + {name: "invalid Native", err: invalidNative.Err(), code: FailureInvalidArgument}, + {name: "Native without its end", err: endless.Err(), interrupted: true}, + {name: "no Native result", err: noNative.Err(), interrupted: true}, + } + for _, test := range cases { + if (test.err == nil) != test.ok || CodeOf(test.err) != test.code || errors.Is(test.err, ErrInterrupted) != test.interrupted { + t.Error(test.name, test.err) + } + } + var failed *Error + if !errors.As(lost.Err(), &failed) || failed.Completion != NativeResponseIncomplete { + t.Error("Native error lost its completion", lost.Err()) + } + if text := lost.Err().Error(); text != "weir: native RESPONSE_INCOMPLETE: UNAVAILABLE: backend unavailable" { + t.Error(text) + } + if CodeOf(errors.New("other")) != 0 || CodeOf(nil) != 0 { + t.Error("code of an error without a Failure") + } +} + +func TestLostConnectionIsInterruptedAndKeepsItsStatus(t *testing.T) { + lost := status.Error(codes.Unavailable, "connection lost") + err := callError(t.Context(), lost) + if !errors.Is(err, ErrInterrupted) || status.Code(err) != codes.Unavailable { + t.Fatal("lost connection not marked", err) + } + other := status.Error(codes.Internal, "server failed") + if err := callError(t.Context(), other); errors.Is(err, ErrInterrupted) { + t.Fatal("another status was marked", err) + } + ended, cancel := context.WithCancel(t.Context()) + cancel() + if err := callError(ended, lost); err != context.Canceled { + t.Fatal("an ended context was marked", err) + } +} diff --git a/examples/mongodb/main.go b/examples/mongodb/main.go new file mode 100644 index 0000000..e9ecf4f --- /dev/null +++ b/examples/mongodb/main.go @@ -0,0 +1,72 @@ +// MongoDB writes, changes, finds and deletes documents of a typed collection. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "time" + + weir "github.com/batchstream/weir-go" + "github.com/batchstream/weir-go/mongodb" + "go.mongodb.org/mongo-driver/v2/bson" +) + +type note struct { + Text string `bson:"text"` + Views int64 `bson:"views"` +} + +func main() { + if err := run(); err != nil { + panic(err) + } +} + +func run() error { + address := flag.String("address", "127.0.0.1:7447", "Weir application address") + store := flag.String("store", "mongo", "Store whose backend is MongoDB") + database := flag.String("database", "weir_example", "existing database") + collection := flag.String("collection", "records", "existing collection") + flag.Parse() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + client, err := weir.New(*address) + if err != nil { + return err + } + defer client.Close() + + db := mongodb.DB{Client: client, Store: *store, Name: *database} + notes := mongodb.Collection[string, note]{DB: db, Name: *collection} + + first := ¬e{Text: "hello"} + if err := notes.Put(ctx, "example-note", first); err != nil { + return err + } + view := bson.D{{Key: "$inc", Value: bson.D{{Key: "views", Value: 1}}}} + if err := notes.Update(ctx, "example-note", view); err != nil { + return err + } + stored, err := notes.Get(ctx, "example-note") + if err != nil { + return err + } + fmt.Printf("get: %s, %d view\n", stored.Text, stored.Views) + + viewed := bson.D{{Key: "views", Value: bson.D{{Key: "$gte", Value: 1}}}} + for found, err := range notes.Find(ctx, viewed) { + if err != nil { + return err + } + fmt.Println("find:", found.Text) + } + if err := notes.Delete(ctx, "example-note"); err != nil { + return err + } + _, err = notes.Get(ctx, "example-note") + fmt.Println("deleted:", errors.Is(err, weir.ErrNotFound)) + return nil +} diff --git a/examples/search/main.go b/examples/search/main.go new file mode 100644 index 0000000..d6d3e5f --- /dev/null +++ b/examples/search/main.go @@ -0,0 +1,73 @@ +// Search writes documents of a typed index, searches them and deletes them. +package main + +import ( + "context" + "flag" + "fmt" + "net/http" + "time" + + weir "github.com/batchstream/weir-go" + "github.com/batchstream/weir-go/search" +) + +type article struct { + Title string `json:"title"` + Topic string `json:"topic"` +} + +func main() { + if err := run(); err != nil { + panic(err) + } +} + +func run() error { + address := flag.String("address", "127.0.0.1:7447", "Weir application address") + store := flag.String("store", "search", "Store whose backend is Elasticsearch or OpenSearch") + index := flag.String("index", "weir_example", "existing index") + flag.Parse() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + client, err := weir.New(*address) + if err != nil { + return err + } + defer client.Close() + + cluster := search.Cluster{Client: client, Store: *store} + articles := search.Index[article]{Cluster: cluster, Name: *index, QueryVisible: true} + + ids := []string{"example-a", "example-b"} + titles := []string{"Rivers", "Weirs"} + for i, id := range ids { + written := &article{Title: titles[i], Topic: "example-water"} + if err := articles.Put(ctx, id, written); err != nil { + return err + } + } + match := map[string]any{"term": map[string]any{"topic.keyword": "example-water"}} + order := []any{map[string]any{"title.keyword": "asc"}} + query := map[string]any{"query": match, "sort": order} + hits, err := articles.Search(ctx, query) + if err != nil { + return err + } + for _, hit := range hits { + fmt.Println("hit:", hit.ID, hit.Document.Title) + } + + health, err := cluster.Do(ctx, http.MethodGet, "/_cluster/health", nil) + if err != nil { + return err + } + fmt.Println("cluster health status:", health.StatusCode) + for _, id := range ids { + if err := articles.Delete(ctx, id); err != nil { + return err + } + } + return nil +} diff --git a/internal/record/record.go b/internal/record/record.go new file mode 100644 index 0000000..412f3d2 --- /dev/null +++ b/internal/record/record.go @@ -0,0 +1,86 @@ +// Package record sends the single-record calls of the mongodb and search +// packages. +package record + +import ( + "context" + "errors" + "fmt" + + weir "github.com/batchstream/weir-go" +) + +// Target is one Store of a Client. +type Target struct { + Client *weir.Client + Store string +} + +var errNoClient = errors.New("weir: no Client") + +// Mutate sends one mutation and returns nil only for APPLIED without a +// Failure. With visible, success also confirms that later queries see it. A +// confirmed result stands even when an RPC error came after it. +func (t Target) Mutate(ctx context.Context, request *weir.MutateRequest, visible bool) error { + if t.Client == nil { + return errNoClient + } + options := []weir.MutateOption{weir.WithRequests(request)} + if visible { + options = append(options, weir.WithQueryVisible()) + } + results, err := t.Client.Mutate(ctx, t.Store, options...) + if err != nil && (len(results) != 1 || results[0] == nil) { + return err + } + if len(results) != 1 { + return fmt.Errorf("weir: %d results for one mutation", len(results)) + } + return weir.MutationErr(results[0]) +} + +// Read reads resources in one call and returns their documents in order, nil +// for a missing record. A failed read fails the whole call. +func (t Target) Read(ctx context.Context, resources []string) ([]*weir.Document, error) { + if t.Client == nil { + return nil, errNoClient + } + documents := make([]*weir.Document, len(resources)) + if len(resources) == 0 { + return documents, nil + } + requests := make([]*weir.ReadRequest, len(resources)) + for i, resource := range resources { + requests[i] = &weir.ReadRequest{Resource: resource} + } + results, err := t.Client.Read(ctx, t.Store, weir.WithRequests(requests...)) + if err != nil { + return nil, err + } + if len(results) != len(resources) { + return nil, fmt.Errorf("weir: %d results for %d reads", len(results), len(resources)) + } + for i, result := range results { + if err := result.Err(); err != nil { + return nil, fmt.Errorf("read %s: %w", resources[i], err) + } + documents[i] = result.Document + } + return documents, nil +} + +// Native sends one Native request and returns its complete result. A complete +// result stands even when an RPC error came after it. +func (t Target) Native(ctx context.Context, request *weir.NativeRequest) (*weir.NativeResult, error) { + if t.Client == nil { + return nil, errNoClient + } + result, err := t.Client.Native(ctx, t.Store, request) + if result != nil && result.Err() == nil { + return result, nil + } + if err != nil { + return nil, err + } + return nil, result.Err() +} diff --git a/internal/storetest/storetest.go b/internal/storetest/storetest.go new file mode 100644 index 0000000..d946ee3 --- /dev/null +++ b/internal/storetest/storetest.go @@ -0,0 +1,166 @@ +// Package storetest serves a fake Weir to the tests of the mongodb and search +// packages. +package storetest + +import ( + "io" + "net" + "sync" + "testing" + + weir "github.com/batchstream/weir-go" + pb "github.com/batchstream/weir-protocol/api/weir/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// Handler answers one command sent to a Store with the events Weir would send. +type Handler func(store string, command *pb.Command) []*pb.Event + +// Lost, in place of an event, ends the stream with UNAVAILABLE as a lost +// connection does. +var Lost = &pb.Event{} + +type server struct { + mu sync.Mutex + handler Handler +} + +func (s *server) Execute(stream grpc.BidiStreamingServer[pb.ExecuteRequests, pb.ExecuteResponses]) error { + // A stream may name its Store only in its first request. + store := "" + for { + message, err := stream.Recv() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + for _, request := range message.Requests { + if request.CancelIndex != 0 { + continue + } + if request.StoreName != "" { + store = request.StoreName + } + s.mu.Lock() + events := s.handler(store, request.Command) + s.mu.Unlock() + for _, event := range events { + if event == Lost { + return status.Error(codes.Unavailable, "connection lost") + } + response := &pb.ExecuteResponse{Index: request.Index, Event: event} + responses := &pb.ExecuteResponses{Responses: []*pb.ExecuteResponse{response}} + if err := stream.Send(responses); err != nil { + return err + } + } + } + } +} + +// Client serves handler on loopback until the test ends and returns a Client +// of it. The handler runs for one command at a time. +func Client(t testing.TB, handler Handler) *weir.Client { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + fake := &server{handler: handler} + service := grpc.NewServer() + pb.RegisterStoreServiceServer(service, fake) + done := make(chan error, 1) + go func() { done <- service.Serve(listener) }() + client, err := weir.New(listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = client.Close() + service.Stop() + <-done + }) + return client +} + +// Read is the event of a read that returned document. +func Read(document *pb.Document) []*pb.Event { + result := &pb.ReadResult{Result: &pb.ReadResult_Document{Document: document}} + return readEvent(result) +} + +// Missing is the event of a read that found no record. +func Missing() []*pb.Event { + missing := &pb.Empty{} + result := &pb.ReadResult{Result: &pb.ReadResult_Missing{Missing: missing}} + return readEvent(result) +} + +// ReadFailure is the event of a failed read. +func ReadFailure(code pb.FailureCode) []*pb.Event { + failure := &pb.Failure{Code: code, Message: "test failure"} + result := &pb.ReadResult{Result: &pb.ReadResult_Failure{Failure: failure}} + return readEvent(result) +} + +func readEvent(result *pb.ReadResult) []*pb.Event { + value := &pb.Event_ReadResult{ReadResult: result} + event := &pb.Event{Value: value} + return []*pb.Event{event} +} + +// Applied is the event of an applied mutation. +func Applied() []*pb.Event { + return Mutation(pb.MutationOutcome_APPLIED, 0) +} + +// Mutation is the event of a mutation result, with a Failure unless code is +// zero. +func Mutation(outcome pb.MutationOutcome, code pb.FailureCode) []*pb.Event { + result := &pb.MutationResult{Outcome: outcome} + if code != 0 { + result.Failure = &pb.Failure{Code: code, Message: "test failure"} + } + value := &pb.Event_MutationResult{MutationResult: result} + event := &pb.Event{Value: value} + return []*pb.Event{event} +} + +// Native is the events of a complete Native response. +func Native(metadata *pb.Document, body []byte) []*pb.Event { + events := NativeStarted(metadata, body) + end := &pb.NativeEnd{Completion: pb.NativeCompletion_RESPONSE_COMPLETE} + value := &pb.Event_NativeEnd{NativeEnd: end} + event := &pb.Event{Value: value} + return append(events, event) +} + +// NativeStarted is the events of a Native response that began, without its +// end. +func NativeStarted(metadata *pb.Document, body []byte) []*pb.Event { + head := &pb.NativeHead{Metadata: metadata} + headValue := &pb.Event_Head{Head: head} + headEvent := &pb.Event{Value: headValue} + events := []*pb.Event{headEvent} + if len(body) > 0 { + chunk := &pb.Event_Chunk{Chunk: body} + chunkEvent := &pb.Event{Value: chunk} + events = append(events, chunkEvent) + } + return events +} + +// NativeIncomplete is the events of a Native response that began and stopped +// before its end with a Failure. +func NativeIncomplete(metadata *pb.Document, body []byte, code pb.FailureCode) []*pb.Event { + events := NativeStarted(metadata, body) + failure := &pb.Failure{Code: code, Message: "test failure"} + end := &pb.NativeEnd{Completion: pb.NativeCompletion_RESPONSE_INCOMPLETE, Failure: failure} + value := &pb.Event_NativeEnd{NativeEnd: end} + event := &pb.Event{Value: value} + return append(events, event) +} diff --git a/mongodb/collection.go b/mongodb/collection.go new file mode 100644 index 0000000..4c4e42b --- /dev/null +++ b/mongodb/collection.go @@ -0,0 +1,306 @@ +package mongodb + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "iter" + "reflect" + "strconv" + + weir "github.com/batchstream/weir-go" + "go.mongodb.org/mongo-driver/v2/bson" +) + +// ID is the type of a record's _id: a string, an integer or an ObjectID. +type ID interface { + ~string | ~int | ~int32 | ~int64 | bson.ObjectID +} + +// Collection holds documents of type T whose _id values have type K. Writes +// take the _id separately; a document may omit its _id field, and one it has +// must equal the id. The collection must exist before its first write. +type Collection[K ID, T any] struct { + DB DB + Name string + // QueryVisible makes each write return only once later queries see it. + QueryVisible bool +} + +// Get returns the document with the id, or weir.ErrNotFound. +func (c Collection[K, T]) Get(ctx context.Context, id K) (*T, error) { + documents, err := c.GetMany(ctx, id) + if err != nil { + return nil, err + } + if documents[0] == nil { + return nil, fmt.Errorf("mongodb: %s %v: %w", c.Name, id, weir.ErrNotFound) + } + return documents[0], nil +} + +// GetMany returns the documents with the ids in one call, in order, with nil +// for each missing one. +func (c Collection[K, T]) GetMany(ctx context.Context, ids ...K) ([]*T, error) { + resources := make([]string, len(ids)) + for i, id := range ids { + segment, _ := key(id) + resources[i] = c.resource(segment) + } + documents, err := c.DB.target().Read(ctx, resources) + if err != nil { + return nil, fmt.Errorf("mongodb: %w", err) + } + values := make([]*T, len(documents)) + for i, document := range documents { + if document == nil { + continue + } + value := new(T) + if err := bson.Unmarshal(document.GetData(), value); err != nil { + return nil, fmt.Errorf("mongodb: decode %s %v: %w", c.Name, ids[i], err) + } + values[i] = value + } + return values, nil +} + +// Create writes a document that must not exist. An existing one, or a unique +// index that rejects it, fails it with FailurePreconditionFailed. +func (c Collection[K, T]) Create(ctx context.Context, id K, document *T) error { + return c.write(ctx, weir.MutationCreate, id, document) +} + +// Put creates or replaces the document. +func (c Collection[K, T]) Put(ctx context.Context, id K, document *T) error { + return c.write(ctx, weir.MutationPut, id, document) +} + +// Replace replaces a document that must exist. A missing one fails it with +// FailurePreconditionFailed. +func (c Collection[K, T]) Replace(ctx context.Context, id K, document *T) error { + return c.write(ctx, weir.MutationReplace, id, document) +} + +// Delete deletes the document. Deleting a missing document succeeds. +func (c Collection[K, T]) Delete(ctx context.Context, id K) error { + segment, _ := key(id) + request := &weir.MutateRequest{Resource: c.resource(segment), Action: weir.MutationDelete} + return c.mutate(ctx, request) +} + +// Update changes fields of a document that must exist with an update document +// of $set, $unset and $inc operators. A missing document fails it with +// FailurePreconditionFailed. +func (c Collection[K, T]) Update(ctx context.Context, id K, update any) error { + data, err := bson.Marshal(update) + if err != nil { + return fmt.Errorf("mongodb: encode update: %w", err) + } + segment, _ := key(id) + expression := &weir.Document{ContentType: updateContentType, Data: data} + request := &weir.MutateRequest{Resource: c.resource(segment), Action: weir.MutationAtomicTransform, BackendExpression: expression} + return c.mutate(ctx, request) +} + +// Transform runs a Lua program atomically against the document. The program +// returns one function, called as function(current, incoming), where current +// is the stored document or nil and incoming is input encoded as BSON, or nil +// when input is nil. Its result replaces the document, and weir.reject(message) +// fails it with FailurePreconditionFailed. See the Lua guide for the rest. +func (c Collection[K, T]) Transform(ctx context.Context, id K, program string, input any) error { + lua := &weir.LuaTransform{Source: []byte(program)} + if input != nil { + data, err := bson.Marshal(input) + if err != nil { + return fmt.Errorf("mongodb: encode transform input: %w", err) + } + lua.Input = &weir.Document{ContentType: bsonContentType, Data: data} + } + segment, _ := key(id) + request := &weir.MutateRequest{Resource: c.resource(segment), Action: weir.MutationAtomicTransform, Lua: lua} + return c.mutate(ctx, request) +} + +// Find yields the documents that match a filter, nil for all. Options are +// further fields of MongoDB's find command, such as sort, projection, skip, +// limit and batchSize. An error ends the sequence. +func (c Collection[K, T]) Find(ctx context.Context, filter any, options ...bson.E) iter.Seq2[*T, error] { + command := bson.D{{Key: "find", Value: c.Name}, {Key: "filter", Value: orEmpty(filter)}} + command = append(command, options...) + values := func(yield func(*T, error) bool) { + for document, err := range c.DB.Cursor(ctx, command) { + if err != nil { + yield(nil, err) + return + } + value := new(T) + if err := bson.Unmarshal(document, value); err != nil { + yield(nil, fmt.Errorf("mongodb: decode %s: %w", c.Name, err)) + return + } + if !yield(value, nil) { + return + } + } + } + return values +} + +// Count returns the number of documents that match a filter, nil for all. It +// counts the matching documents; for a cheap estimate of a whole collection, +// run the count command with DB.Command. +func (c Collection[K, T]) Count(ctx context.Context, filter any) (int64, error) { + match := bson.D{{Key: "$match", Value: orEmpty(filter)}} + total := bson.D{{Key: "$count", Value: "n"}} + command := bson.D{ + {Key: "aggregate", Value: c.Name}, + {Key: "pipeline", Value: bson.A{match, total}}, + {Key: "cursor", Value: bson.D{}}, + } + var count int64 + for document, err := range c.DB.Cursor(ctx, command) { + if err != nil { + return 0, err + } + count, _ = document.Lookup("n").AsInt64OK() + } + return count, nil +} + +func (c Collection[K, T]) write(ctx context.Context, action weir.MutationAction, id K, document *T) error { + if document == nil { + return errors.New("mongodb: document is nil") + } + segment, stored := key(id) + encoded, err := recordDocument(stored, document) + if err != nil { + return fmt.Errorf("mongodb: encode %s %v: %w", c.Name, id, err) + } + request := &weir.MutateRequest{Resource: c.resource(segment), Action: action, Document: encoded} + return c.mutate(ctx, request) +} + +func (c Collection[K, T]) mutate(ctx context.Context, request *weir.MutateRequest) error { + err := c.DB.target().Mutate(ctx, request, c.QueryVisible) + if err != nil { + return fmt.Errorf("mongodb: %s: %w", request.Resource, err) + } + return nil +} + +// resource addresses one record: database/collection/key. +func (c Collection[K, T]) resource(segment string) string { + resource := weir.EncodeSegment(c.DB.Name) + "/" + weir.EncodeSegment(c.Name) + "/" + weir.EncodeSegment(segment) + return resource +} + +// key returns the resource key of an id, s:, i: or oid:, and the _id value +// Weir stores for it: a string, an int64 or an ObjectID. +func key[K ID](id K) (string, any) { + value := reflect.ValueOf(id) + switch value.Kind() { + case reflect.String: + text := value.String() + return "s:" + text, text + case reflect.Int, reflect.Int32, reflect.Int64: + number := value.Int() + return "i:" + strconv.FormatInt(number, 10), number + } + object := any(id).(bson.ObjectID) + return "oid:" + object.Hex(), object +} + +// recordDocument encodes value as BSON whose first field is an _id equal to +// id, as Weir requires. It moves an _id the value has to the front, keeping +// its bytes, and adds one when it has none or leaves it zero. +func recordDocument(id any, value any) (*weir.Document, error) { + raw, err := bson.Marshal(value) + if err != nil { + return nil, err + } + elements, err := bson.Raw(raw).Elements() + if err != nil { + return nil, err + } + var first []byte + for i, element := range elements { + if element.Key() != "_id" || zeroID(element.Value()) { + continue + } + if !sameID(element.Value(), id) { + return nil, fmt.Errorf("document _id %v differs from %v", element.Value(), id) + } + if i == 0 { + document := &weir.Document{ContentType: bsonContentType, Data: raw} + return document, nil + } + first = element + } + if first == nil { + kind, data, err := bson.MarshalValue(id) + if err != nil { + return nil, err + } + first = append([]byte{byte(kind)}, "_id\x00"...) + first = append(first, data...) + } + data := make([]byte, 4, len(raw)+len(first)) + data = append(data, first...) + for _, element := range elements { + if element.Key() != "_id" { + data = append(data, element...) + } + } + data = append(data, 0) + binary.LittleEndian.PutUint32(data, uint32(len(data))) + document := &weir.Document{ContentType: bsonContentType, Data: data} + return document, nil +} + +// zeroID reports an _id a document left at its zero value or null, as a +// struct field without omitempty does, which counts as no _id. +func zeroID(value bson.RawValue) bool { + switch value.Type { + case bson.TypeNull: + return true + case bson.TypeString: + return value.StringValue() == "" + case bson.TypeInt32: + return value.Int32() == 0 + case bson.TypeInt64: + return value.Int64() == 0 + case bson.TypeObjectID: + return value.ObjectID().IsZero() + } + return false +} + +// sameID reports whether a stored _id matches an id as Weir compares them: a +// string, an int32 or int64 integer, or an ObjectID. +func sameID(stored bson.RawValue, id any) bool { + switch want := id.(type) { + case string: + text, ok := stored.StringValueOK() + return ok && text == want + case int64: + if number, ok := stored.Int32OK(); ok { + return int64(number) == want + } + number, ok := stored.Int64OK() + return ok && number == want + case bson.ObjectID: + object, ok := stored.ObjectIDOK() + return ok && object == want + } + return false +} + +func orEmpty(filter any) any { + if filter == nil { + empty := bson.D{} + return empty + } + return filter +} diff --git a/mongodb/integration_test.go b/mongodb/integration_test.go new file mode 100644 index 0000000..4e0ab49 --- /dev/null +++ b/mongodb/integration_test.go @@ -0,0 +1,196 @@ +//go:build integration + +package mongodb_test + +import ( + "context" + "errors" + "fmt" + "os" + "strings" + "testing" + "time" + + weir "github.com/batchstream/weir-go" + "github.com/batchstream/weir-go/mongodb" + "go.mongodb.org/mongo-driver/v2/bson" +) + +type person struct { + ID string `bson:"_id,omitempty"` + Name string `bson:"name"` + Visits int64 `bson:"visits"` +} + +// collection returns the pre-created disposable collection named by +// WEIR_MONGO_RESOURCE, database/collection, behind WEIR_MONGO_STORE. +func collection(t *testing.T) (mongodb.DB, string) { + t.Helper() + address, store, resource := os.Getenv("WEIR_ADDRESS"), os.Getenv("WEIR_MONGO_STORE"), os.Getenv("WEIR_MONGO_RESOURCE") + database, name, found := strings.Cut(resource, "/") + if address == "" || store == "" || !found { + t.Skip("requires WEIR_ADDRESS, WEIR_MONGO_STORE and WEIR_MONGO_RESOURCE") + } + client, err := weir.New(address) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = client.Close() }) + db := mongodb.DB{Client: client, Store: store, Name: database} + return db, name +} + +func TestCollectionLifecycle(t *testing.T) { + db, name := collection(t) + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + people := mongodb.Collection[string, person]{DB: db, Name: name, QueryVisible: true} + prefix := fmt.Sprintf("sdk-mongodb-%d-", time.Now().UnixNano()) + ids := []string{prefix + "a", prefix + "b", prefix + "c", prefix + "d", prefix + "e"} + t.Cleanup(func() { + cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + for _, id := range ids { + _ = people.Delete(cleanup, id) + } + }) + + ada := &person{Name: "Ada"} + if err := people.Create(ctx, ids[0], ada); err != nil { + t.Fatal(err) + } + if err := people.Create(ctx, ids[0], ada); weir.CodeOf(err) != weir.FailurePreconditionFailed { + t.Fatal("second create was not refused", err) + } + got, err := people.Get(ctx, ids[0]) + if err != nil || got.ID != ids[0] || got.Name != "Ada" { + t.Fatal("wrong document", got, err) + } + missing := &person{Name: "Nobody"} + if err := people.Replace(ctx, ids[1], missing); weir.CodeOf(err) != weir.FailurePreconditionFailed { + t.Fatal("replace of a missing document was not refused", err) + } + visit := bson.D{{Key: "$inc", Value: bson.D{{Key: "visits", Value: int64(2)}}}} + if err := people.Update(ctx, ids[0], visit); err != nil { + t.Fatal(err) + } + const rename = `return function(current, incoming) + if current == nil then return weir.reject("no such person") end + current.name = incoming.name + return current +end` + renamed := bson.D{{Key: "name", Value: "Ada L."}} + err = people.Transform(ctx, ids[0], rename, renamed) + if err != nil { + t.Fatal(err) + } + err = people.Transform(ctx, ids[1], rename, renamed) + if weir.CodeOf(err) != weir.FailurePreconditionFailed || !strings.Contains(err.Error(), "no such person") { + t.Fatal("rejection lost", err) + } + got, err = people.Get(ctx, ids[0]) + if err != nil || got.Name != "Ada L." || got.Visits != 2 { + t.Fatal("update or transform lost", got, err) + } + + for _, id := range ids[1:] { + other := &person{ID: id, Name: "Other"} + if err := people.Put(ctx, id, other); err != nil { + t.Fatal(err) + } + } + many, err := people.GetMany(ctx, ids[4], prefix+"none", ids[0]) + if err != nil || many[0].ID != ids[4] || many[1] != nil || many[2].ID != ids[0] { + t.Fatal("wrong documents", many, err) + } + mine := bson.D{{Key: "_id", Value: bson.D{{Key: "$regex", Value: "^" + prefix}}}} + order := bson.E{Key: "sort", Value: bson.D{{Key: "_id", Value: -1}}} + small := bson.E{Key: "batchSize", Value: 2} + var found []string + for found1, err := range people.Find(ctx, mine, order, small) { + if err != nil { + t.Fatal(err) + } + found = append(found, found1.ID) + } + if len(found) != 5 || found[0] != ids[4] || found[4] != ids[0] { + t.Fatal("find did not follow its cursor", found) + } + for first, err := range people.Find(ctx, mine, small) { + if err != nil || first.ID != ids[0] { + t.Fatal(first, err) + } + break + } + count, err := people.Count(ctx, mine) + if err != nil || count != 5 { + t.Fatal("wrong count", count, err) + } + + for _, id := range ids { + if err := people.Delete(ctx, id); err != nil { + t.Fatal(err) + } + } + if _, err := people.Get(ctx, ids[0]); !errors.Is(err, weir.ErrNotFound) { + t.Fatal("deleted document still found", err) + } + if err := people.Delete(ctx, ids[0]); err != nil { + t.Fatal("deleting a missing document failed", err) + } +} + +func TestCollectionKeysOfEveryKind(t *testing.T) { + db, name := collection(t) + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + numbers := mongodb.Collection[int64, bson.M]{DB: db, Name: name} + number := time.Now().UnixNano() + value := bson.M{"kind": "number"} + if err := numbers.Put(ctx, number, &value); err != nil { + t.Fatal(err) + } + defer numbers.Delete(context.WithoutCancel(ctx), number) + got, err := numbers.Get(ctx, number) + if err != nil || (*got)["_id"] != number { + t.Fatal("integer key lost", got, err) + } + objects := mongodb.Collection[bson.ObjectID, bson.M]{DB: db, Name: name} + object := bson.NewObjectID() + value = bson.M{"kind": "object"} + if err := objects.Put(ctx, object, &value); err != nil { + t.Fatal(err) + } + defer objects.Delete(context.WithoutCancel(ctx), object) + got, err = objects.Get(ctx, object) + if err != nil || (*got)["_id"] != object { + t.Fatal("ObjectID key lost", got, err) + } +} + +func TestCommandReportsMongoDBFailures(t *testing.T) { + db, name := collection(t) + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + ping := bson.D{{Key: "ping", Value: 1}} + if _, err := db.Command(ctx, ping); err != nil { + t.Fatal(err) + } + invalid := bson.D{{Key: "find", Value: name}, {Key: "filter", Value: 1}} + var failed *mongodb.CommandError + if _, err := db.Command(ctx, invalid); !errors.As(err, &failed) || failed.Code == 0 || failed.Name == "" { + t.Fatal("ok: 0 not reported", err) + } + id := fmt.Sprintf("sdk-mongodb-dup-%d", time.Now().UnixNano()) + document := bson.D{{Key: "_id", Value: id}} + insert := bson.D{{Key: "insert", Value: name}, {Key: "documents", Value: bson.A{document, document}}} + _, err := db.Command(ctx, insert) + defer func() { + remove := bson.D{{Key: "q", Value: document}, {Key: "limit", Value: 0}} + cleanup := bson.D{{Key: "delete", Value: name}, {Key: "deletes", Value: bson.A{remove}}} + _, _ = db.Command(context.WithoutCancel(ctx), cleanup) + }() + if !errors.As(err, &failed) || failed.Code != 11000 || failed.Reply.Lookup("n").Int32() != 1 { + t.Fatal("write error not reported", err) + } +} diff --git a/mongodb/mongodb.go b/mongodb/mongodb.go new file mode 100644 index 0000000..a374673 --- /dev/null +++ b/mongodb/mongodb.go @@ -0,0 +1,233 @@ +// Package mongodb reads and writes MongoDB documents through a Weir Store +// whose backend is MongoDB. +// +// A Collection reads and writes documents by _id and runs find and count +// queries. A DB runs any other command: +// +// db := mongodb.DB{Client: client, Store: "mongo", Name: "shop"} +// users := mongodb.Collection[string, User]{DB: db, Name: "users"} +// +// ada := &User{Name: "Ada"} +// err := users.Put(ctx, "ada", ada) +// user, err := users.Get(ctx, "ada") +// +// active := bson.D{{Key: "active", Value: true}} +// for user, err := range users.Find(ctx, active) { +// ... +// } +// +// ping := bson.D{{Key: "ping", Value: 1}} +// reply, err := db.Command(ctx, ping) +// +// Set a deadline on every context. A failed record is a *weir.Error: check it +// with weir.CodeOf, and with errors.Is against weir.ErrNotFound and +// weir.ErrInterrupted. The package keeps no state of its own; DB and +// Collection are plain values that are safe to copy and share. +package mongodb + +import ( + "context" + "fmt" + "iter" + "strings" + "time" + + weir "github.com/batchstream/weir-go" + "github.com/batchstream/weir-go/internal/record" + "go.mongodb.org/mongo-driver/v2/bson" +) + +const ( + bsonContentType = "application/bson" + updateContentType = "application/vnd.weir.mongodb-update.v1+bson" + + // killTimeout bounds the killCursors sent after a loop stops early. + killTimeout = 5 * time.Second +) + +// DB is a MongoDB database behind a Weir Store. +type DB struct { + Client *weir.Client + // Store names a Weir Store whose backend is MongoDB. + Store string + // Name is the database name. + Name string +} + +// CommandError is a command MongoDB answered with ok: 0, a write error or a +// write concern error. For write errors it describes the first one. +type CommandError struct { + Code int32 + Name string + Message string + // Reply is the complete reply. + Reply bson.Raw +} + +func (e *CommandError) Error() string { + if e.Name == "" { + return fmt.Sprintf("mongodb: command failed: code %d: %s", e.Code, e.Message) + } + return fmt.Sprintf("mongodb: command failed: %s (%d): %s", e.Name, e.Code, e.Message) +} + +// Command runs one command in the database and returns its reply. A reply +// with ok: 0, writeErrors or a writeConcernError comes with a *CommandError. +// A command is sent again only when a server refused it before it began. One +// a lost connection or an unavailable backend cut short reports +// weir.ErrInterrupted and may have run; send it again only when that is safe. +func (db DB) Command(ctx context.Context, command any) (bson.Raw, error) { + data, err := bson.Marshal(command) + if err != nil { + return nil, fmt.Errorf("mongodb: encode command: %w", err) + } + document := &weir.Document{ContentType: bsonContentType, Data: data} + request := &weir.NativeRequest{Resource: db.resource(), Request: document} + result, err := db.target().Native(ctx, request) + if err != nil { + return nil, err + } + reply := bson.Raw(result.Body) + if err := reply.Validate(); err != nil { + return nil, fmt.Errorf("mongodb: invalid reply: %w", err) + } + if failed := commandError(reply); failed != nil { + return reply, failed + } + return reply, nil +} + +// Cursor runs a command that opens a cursor, such as find, aggregate or +// listIndexes, and yields its documents. It sends getMore until the cursor +// ends, and kills the cursor when the loop stops early or fails. An error +// ends the sequence. +func (db DB) Cursor(ctx context.Context, command any) iter.Seq2[bson.Raw, error] { + documents := func(yield func(bson.Raw, error) bool) { + reply, err := db.Command(ctx, command) + if err != nil { + yield(nil, err) + return + } + page, err := readPage(reply) + if err != nil { + yield(nil, err) + return + } + space := parseNamespace(page.Cursor.Namespace) + batch, cursor := page.Cursor.FirstBatch, page.Cursor.ID + for { + for _, document := range batch { + if !yield(document, nil) { + db.killCursor(ctx, space, cursor) + return + } + } + if cursor == 0 { + return + } + getMore := bson.D{{Key: "getMore", Value: cursor}, {Key: "collection", Value: space.collection}, {Key: "$db", Value: space.database}} + reply, err := db.Command(ctx, getMore) + var next cursorPage + if err == nil { + next, err = readPage(reply) + } + if err != nil { + db.killCursor(ctx, space, cursor) + yield(nil, err) + return + } + batch, cursor = next.Cursor.NextBatch, next.Cursor.ID + } + } + return documents +} + +// namespace names the collection of a cursor. Commands that follow the cursor +// name its database, as the command that opened it may have named one with +// $db. +type namespace struct { + database string + collection string +} + +// parseNamespace splits database.collection; database names hold no dot. +func parseNamespace(text string) namespace { + database, collection, _ := strings.Cut(text, ".") + space := namespace{database: database, collection: collection} + return space +} + +// cursorPage is the cursor of a find, aggregate or getMore reply. +type cursorPage struct { + Cursor *struct { + ID int64 `bson:"id"` + Namespace string `bson:"ns"` + FirstBatch []bson.Raw `bson:"firstBatch"` + NextBatch []bson.Raw `bson:"nextBatch"` + } `bson:"cursor"` +} + +func readPage(reply bson.Raw) (cursorPage, error) { + var page cursorPage + if err := bson.Unmarshal(reply, &page); err != nil { + return page, fmt.Errorf("mongodb: invalid cursor reply: %w", err) + } + if page.Cursor == nil { + return page, fmt.Errorf("mongodb: reply has no cursor") + } + return page, nil +} + +// killCursor releases a cursor a loop left open. The server also times idle +// cursors out, so a failure here is ignored. +func (db DB) killCursor(ctx context.Context, space namespace, cursor int64) { + if cursor == 0 { + return + } + cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), killTimeout) + defer cancel() + command := bson.D{{Key: "killCursors", Value: space.collection}, {Key: "cursors", Value: bson.A{cursor}}, {Key: "$db", Value: space.database}} + _, _ = db.Command(cleanup, command) +} + +func (db DB) target() record.Target { + target := record.Target{Client: db.Client, Store: db.Store} + return target +} + +// resource addresses the database. An empty Name leaves the database to a +// $db field in each command. +func (db DB) resource() string { + if db.Name == "" { + return "" + } + return weir.EncodeSegment(db.Name) +} + +// commandError returns the failure a reply reports, or nil. +func commandError(reply bson.Raw) *CommandError { + ok, _ := reply.Lookup("ok").AsFloat64OK() + if ok != 1 { + return replyError(reply, reply) + } + if errs, found := reply.Lookup("writeErrors").ArrayOK(); found { + values, _ := errs.Values() + if len(values) > 0 { + first, _ := values[0].DocumentOK() + return replyError(reply, first) + } + } + if concern, found := reply.Lookup("writeConcernError").DocumentOK(); found { + return replyError(reply, concern) + } + return nil +} + +// replyError describes the failure in part of a reply. +func replyError(reply, part bson.Raw) *CommandError { + code, _ := part.Lookup("code").AsInt64OK() + name, _ := part.Lookup("codeName").StringValueOK() + message, _ := part.Lookup("errmsg").StringValueOK() + failed := &CommandError{Code: int32(code), Name: name, Message: message, Reply: reply} + return failed +} diff --git a/mongodb/mongodb_test.go b/mongodb/mongodb_test.go new file mode 100644 index 0000000..5d6389f --- /dev/null +++ b/mongodb/mongodb_test.go @@ -0,0 +1,444 @@ +package mongodb + +import ( + "bytes" + "errors" + "slices" + "testing" + + weir "github.com/batchstream/weir-go" + "github.com/batchstream/weir-go/internal/storetest" + pb "github.com/batchstream/weir-protocol/api/weir/v1" + "go.mongodb.org/mongo-driver/v2/bson" +) + +type user struct { + ID string `bson:"_id,omitempty"` + Name string `bson:"name"` +} + +// recorder answers commands with a handler and keeps every command it saw. +type recorder struct { + t *testing.T + commands []*pb.Command + stores []string + answer func(command *pb.Command) []*pb.Event +} + +func (r *recorder) db() DB { + handler := func(store string, command *pb.Command) []*pb.Event { + r.commands = append(r.commands, command) + r.stores = append(r.stores, store) + return r.answer(command) + } + client := storetest.Client(r.t, handler) + db := DB{Client: client, Store: "mongo", Name: "app"} + return db +} + +func (r *recorder) users() Collection[string, user] { + users := Collection[string, user]{DB: r.db(), Name: "users"} + return users +} + +func bsonDocument(t *testing.T, value any) *pb.Document { + t.Helper() + data, err := bson.Marshal(value) + if err != nil { + t.Fatal(err) + } + document := &pb.Document{ContentType: bsonContentType, Data: data} + return document +} + +// command decodes a Native MongoDB command. +func command(t *testing.T, sent *pb.Command) bson.D { + t.Helper() + var decoded bson.D + if err := bson.Unmarshal(sent.GetNative().GetRequest().GetData(), &decoded); err != nil { + t.Fatal(err) + } + return decoded +} + +func TestWritesAddressTheRecordAndPutItsIDFirst(t *testing.T) { + r := &recorder{t: t, answer: func(*pb.Command) []*pb.Event { return storetest.Applied() }} + users := r.users() + alice := &user{Name: "Alice"} + if err := users.Put(t.Context(), "alice/1", alice); err != nil { + t.Fatal(err) + } + put := r.commands[0].GetMutate() + if put.GetResource() != "app/users/s:alice%2F1" || r.stores[0] != "mongo" { + t.Fatal("wrong target", put.GetResource(), r.stores[0]) + } + var stored bson.D + if err := bson.Unmarshal(put.GetPut().GetData(), &stored); err != nil { + t.Fatal(err) + } + if stored[0].Key != "_id" || stored[0].Value != "alice/1" || stored[1].Key != "name" { + t.Fatal("_id is not first", stored) + } + if put.GetOptions().GetVisibility() == pb.MutationVisibility_QUERY_VISIBLE { + t.Fatal("visibility requested without QueryVisible") + } + + users.QueryVisible = true + if err := users.Delete(t.Context(), "alice/1"); err != nil { + t.Fatal(err) + } + remove := r.commands[1].GetMutate() + if remove.GetDelete() == nil || remove.GetOptions().GetVisibility() != pb.MutationVisibility_QUERY_VISIBLE { + t.Fatal("delete did not wait for visibility", remove) + } +} + +func TestIDKindsSelectTheirKeys(t *testing.T) { + object := bson.NewObjectID() + type named string + cases := []struct { + segment string + stored any + want string + }{ + {want: "s:a"}, + {want: "s:n"}, + {want: "i:-7"}, + {want: "i:7"}, + {want: "oid:" + object.Hex()}, + } + cases[0].segment, cases[0].stored = key("a") + cases[1].segment, cases[1].stored = key(named("n")) + cases[2].segment, cases[2].stored = key(-7) + cases[3].segment, cases[3].stored = key(int32(7)) + cases[4].segment, cases[4].stored = key(object) + for _, c := range cases { + if c.segment != c.want { + t.Fatal("wrong key", c.segment, c.want) + } + } + if cases[1].stored != "n" || cases[2].stored != int64(-7) || cases[4].stored != object { + t.Fatal("wrong stored _id", cases) + } +} + +func TestRecordDocumentKeepsOrMovesAnEqualID(t *testing.T) { + first := bson.D{{Key: "_id", Value: int32(5)}, {Key: "n", Value: 1}} + data, err := bson.Marshal(first) + if err != nil { + t.Fatal(err) + } + kept, err := recordDocument(int64(5), first) + if err != nil || !bytes.Equal(kept.Data, data) { + t.Fatal("a leading equal _id was re-encoded", err) + } + later := bson.D{{Key: "n", Value: 1}, {Key: "_id", Value: int32(5)}} + moved, err := recordDocument(int64(5), later) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(moved.Data, data) { + t.Fatal("a later _id was not moved first with its bytes", bson.Raw(moved.Data)) + } + other := bson.D{{Key: "_id", Value: "b"}} + if _, err := recordDocument("a", other); err == nil { + t.Fatal("a different _id was accepted") + } + object := bson.NewObjectID() + missing := bson.D{{Key: "n", Value: 1}} + added, err := recordDocument(object, missing) + if err != nil { + t.Fatal(err) + } + if got := bson.Raw(added.Data).Lookup("_id").ObjectID(); got != object { + t.Fatal("ObjectID _id was not added", got) + } + unset := struct { + ID string `bson:"_id"` + Name string `bson:"name"` + }{Name: "Ada"} + filled, err := recordDocument("ada", unset) + if err != nil { + t.Fatal("a zero _id was taken as a different one", err) + } + if got := bson.Raw(filled.Data).Lookup("_id").StringValue(); got != "ada" { + t.Fatal("a zero _id was not replaced", bson.Raw(filled.Data)) + } +} + +func TestGetDecodesAndReportsMissingRecords(t *testing.T) { + alice := &user{ID: "alice", Name: "Alice"} + r := &recorder{t: t} + r.answer = func(command *pb.Command) []*pb.Event { + if command.GetRead().GetResource() == "app/users/s:alice" { + return storetest.Read(bsonDocument(t, alice)) + } + return storetest.Missing() + } + users := r.users() + got, err := users.Get(t.Context(), "alice") + if err != nil || *got != *alice { + t.Fatal("wrong document", got, err) + } + _, err = users.Get(t.Context(), "bob") + if !errors.Is(err, weir.ErrNotFound) { + t.Fatal("missing record not reported", err) + } + many, err := users.GetMany(t.Context(), "bob", "alice", "bob") + if err != nil || many[0] != nil || *many[1] != *alice || many[2] != nil { + t.Fatal("wrong documents", many, err) + } + empty, err := users.GetMany(t.Context()) + if err != nil || len(empty) != 0 { + t.Fatal("empty read", empty, err) + } +} + +func TestFailedRecordsReportTheirCode(t *testing.T) { + r := &recorder{t: t} + r.answer = func(command *pb.Command) []*pb.Event { + if command.GetRead() != nil { + return storetest.ReadFailure(pb.FailureCode_TARGET_NOT_FOUND) + } + return storetest.Mutation(pb.MutationOutcome_NOT_APPLIED, pb.FailureCode_PRECONDITION_FAILED) + } + users := r.users() + alice := &user{Name: "Alice"} + err := users.Create(t.Context(), "alice", alice) + var failed *weir.Error + if !errors.As(err, &failed) || failed.Outcome != weir.MutationNotApplied || weir.CodeOf(err) != weir.FailurePreconditionFailed { + t.Fatal("wrong create error", err) + } + _, err = users.Get(t.Context(), "alice") + if weir.CodeOf(err) != weir.FailureTargetNotFound || errors.Is(err, weir.ErrNotFound) { + t.Fatal("wrong read error", err) + } + if err := users.Put(t.Context(), "alice", nil); err == nil { + t.Fatal("nil document accepted") + } +} + +func TestUpdateAndTransformSendTheirPayloads(t *testing.T) { + r := &recorder{t: t, answer: func(*pb.Command) []*pb.Event { return storetest.Applied() }} + counters := Collection[int64, bson.M]{DB: r.db(), Name: "counters"} + increment := bson.D{{Key: "$inc", Value: bson.D{{Key: "n", Value: 1}}}} + if err := counters.Update(t.Context(), 9, increment); err != nil { + t.Fatal(err) + } + expression := r.commands[0].GetMutate().GetAtomicTransform().GetBackendExpression() + if r.commands[0].GetMutate().GetResource() != "app/counters/i:9" || expression.GetContentType() != updateContentType { + t.Fatal("wrong update", r.commands[0]) + } + const program = "return function(current, incoming) return incoming end" + input := bson.D{{Key: "n", Value: 2}} + err := counters.Transform(t.Context(), 9, program, input) + if err != nil { + t.Fatal(err) + } + lua := r.commands[1].GetMutate().GetAtomicTransform().GetLua() + if string(lua.GetSource()) != program || lua.GetInput().GetContentType() != bsonContentType { + t.Fatal("wrong transform", lua) + } + err = counters.Transform(t.Context(), 9, program, nil) + if err != nil { + t.Fatal(err) + } + if r.commands[2].GetMutate().GetAtomicTransform().GetLua().GetInput() != nil { + t.Fatal("nil input was sent") + } +} + +func TestCommandReportsFailedReplies(t *testing.T) { + replies := []bson.D{ + {{Key: "ok", Value: 1.0}, {Key: "n", Value: 3}}, + {{Key: "ok", Value: 0.0}, {Key: "code", Value: 26}, {Key: "codeName", Value: "NamespaceNotFound"}, {Key: "errmsg", Value: "ns not found"}}, + {{Key: "ok", Value: 1.0}, {Key: "writeErrors", Value: bson.A{bson.D{{Key: "index", Value: 0}, {Key: "code", Value: 11000}, {Key: "errmsg", Value: "duplicate key"}}}}}, + {{Key: "ok", Value: 1.0}, {Key: "writeConcernError", Value: bson.D{{Key: "code", Value: 64}, {Key: "errmsg", Value: "waiting for replication timed out"}}}}, + } + next := 0 + r := &recorder{t: t} + r.answer = func(*pb.Command) []*pb.Event { + reply := replies[next] + next++ + return storetest.Native(nil, bsonDocument(t, reply).Data) + } + db := r.db() + ping := bson.D{{Key: "ping", Value: 1}} + reply, err := db.Command(t.Context(), ping) + if err != nil || reply.Lookup("n").Int32() != 3 { + t.Fatal("wrong reply", reply, err) + } + if r.commands[0].GetNative().GetResource() != "app" { + t.Fatal("wrong database", r.commands[0]) + } + wantCodes := []int32{26, 11000, 64} + for _, code := range wantCodes { + reply, err := db.Command(t.Context(), ping) + var failed *CommandError + if !errors.As(err, &failed) || failed.Code != code || reply == nil { + t.Fatal("failure not reported", code, err) + } + } +} + +func TestCommandKeepsACompleteReplyDespiteALaterLostConnection(t *testing.T) { + ok := bson.D{{Key: "ok", Value: 1.0}} + lost := false + r := &recorder{t: t} + r.answer = func(*pb.Command) []*pb.Event { + if lost { + return []*pb.Event{storetest.Lost} + } + return append(storetest.Native(nil, bsonDocument(t, ok).Data), storetest.Lost) + } + db := r.db() + ping := bson.D{{Key: "ping", Value: 1}} + reply, err := db.Command(t.Context(), ping) + if err != nil || reply.Lookup("ok").Double() != 1 { + t.Fatal("complete reply was discarded", reply, err) + } + lost = true + if _, err := db.Command(t.Context(), ping); !errors.Is(err, weir.ErrInterrupted) { + t.Fatal("lost connection not interrupted", err) + } +} + +func TestCommandInterruptedByAnUnavailableBackend(t *testing.T) { + code := pb.FailureCode_UNAVAILABLE + r := &recorder{t: t, answer: func(*pb.Command) []*pb.Event { return storetest.NativeIncomplete(nil, []byte{1}, code) }} + db := r.db() + ping := bson.D{{Key: "ping", Value: 1}} + _, err := db.Command(t.Context(), ping) + if !errors.Is(err, weir.ErrInterrupted) || weir.CodeOf(err) != weir.FailureUnavailable { + t.Fatal("unavailable backend did not interrupt", err) + } + code = pb.FailureCode_INVALID_ARGUMENT + _, err = db.Command(t.Context(), ping) + if err == nil || errors.Is(err, weir.ErrInterrupted) { + t.Fatal("a failure of the request itself was interrupted", err) + } +} + +// cursorServer answers find, aggregate and getMore from batches of names and +// records the cursors it was asked to kill. +type cursorServer struct { + t *testing.T + batches [][]string + failAt int + killed []bson.D +} + +func (s *cursorServer) answer(sent *pb.Command) []*pb.Event { + request := command(s.t, sent) + page := 0 + field := "firstBatch" + switch request[0].Key { + case "killCursors": + s.killed = append(s.killed, request) + reply := bson.D{{Key: "ok", Value: 1.0}} + return storetest.Native(nil, bsonDocument(s.t, reply).Data) + case "getMore": + page = int(request[0].Value.(int64)) + field = "nextBatch" + if page == s.failAt { + reply := bson.D{{Key: "ok", Value: 0.0}, {Key: "code", Value: 43}, {Key: "errmsg", Value: "cursor not found"}} + return storetest.Native(nil, bsonDocument(s.t, reply).Data) + } + } + documents := bson.A{} + for _, name := range s.batches[page] { + document := bson.D{{Key: "_id", Value: name}, {Key: "name", Value: name}} + documents = append(documents, document) + } + cursor := int64(page + 1) + if page+1 == len(s.batches) { + cursor = 0 + } + page0 := bson.D{{Key: "id", Value: cursor}, {Key: "ns", Value: "app.users"}, {Key: field, Value: documents}} + reply := bson.D{{Key: "cursor", Value: page0}, {Key: "ok", Value: 1.0}} + return storetest.Native(nil, bsonDocument(s.t, reply).Data) +} + +func TestFindFollowsTheCursorAndKillsAnAbandonedOne(t *testing.T) { + cursors := &cursorServer{t: t, batches: [][]string{{"a", "b"}, {"c"}, {"d"}}, failAt: -1} + r := &recorder{t: t, answer: cursors.answer} + users := r.users() + filter := bson.D{{Key: "name", Value: bson.D{{Key: "$gte", Value: "a"}}}} + order := bson.E{Key: "sort", Value: bson.D{{Key: "_id", Value: 1}}} + var names []string + for found, err := range users.Find(t.Context(), filter, order) { + if err != nil { + t.Fatal(err) + } + names = append(names, found.Name) + } + if !slices.Equal(names, []string{"a", "b", "c", "d"}) || len(cursors.killed) != 0 { + t.Fatal("wrong documents", names, cursors.killed) + } + find := command(t, r.commands[0]) + if find[0].Key != "find" || find[0].Value != "users" || find[2].Key != "sort" { + t.Fatal("wrong find", find) + } + getMore := command(t, r.commands[1]) + if getMore[1].Value != "users" || getMore[2].Key != "$db" || getMore[2].Value != "app" { + t.Fatal("getMore named the wrong collection", getMore) + } + + for found, err := range users.Find(t.Context(), nil) { + if err != nil || found.Name != "a" { + t.Fatal(found, err) + } + break + } + if len(cursors.killed) != 1 || cursors.killed[0][1].Value.(bson.A)[0] != int64(1) || cursors.killed[0][2].Value != "app" { + t.Fatal("abandoned cursor not killed", cursors.killed) + } +} + +func TestFailedGetMoreEndsTheSequenceAndKillsTheCursor(t *testing.T) { + cursors := &cursorServer{t: t, batches: [][]string{{"a"}, {"b"}}, failAt: 1} + r := &recorder{t: t, answer: cursors.answer} + ping := bson.D{{Key: "find", Value: "users"}} + var seen int + var failure error + for _, err := range r.db().Cursor(t.Context(), ping) { + if err != nil { + failure = err + continue + } + seen++ + } + var failed *CommandError + if seen != 1 || !errors.As(failure, &failed) || len(cursors.killed) != 1 { + t.Fatal("failed getMore not reported", seen, failure, cursors.killed) + } +} + +func TestCountReturnsTheMatchingDocuments(t *testing.T) { + r := &recorder{t: t} + empty := false + r.answer = func(*pb.Command) []*pb.Event { + documents := bson.A{bson.D{{Key: "n", Value: int32(42)}}} + if empty { + documents = bson.A{} + } + cursor := bson.D{{Key: "id", Value: int64(0)}, {Key: "ns", Value: "app.users"}, {Key: "firstBatch", Value: documents}} + reply := bson.D{{Key: "cursor", Value: cursor}, {Key: "ok", Value: 1.0}} + return storetest.Native(nil, bsonDocument(t, reply).Data) + } + users := r.users() + filter := bson.D{{Key: "name", Value: "a"}} + count, err := users.Count(t.Context(), filter) + if err != nil || count != 42 { + t.Fatal("wrong count", count, err) + } + aggregate := command(t, r.commands[0]) + if aggregate[0].Key != "aggregate" || aggregate[0].Value != "users" { + t.Fatal("wrong aggregate", aggregate) + } + empty = true + count, err = users.Count(t.Context(), nil) + if err != nil || count != 0 { + t.Fatal("wrong empty count", count, err) + } +} diff --git a/native.go b/native.go index 7f27f10..4af6b2b 100644 --- a/native.go +++ b/native.go @@ -25,7 +25,7 @@ func (c *Client) Native(ctx context.Context, store string, request *NativeReques } call := nativeCall{store: store, request: request, chunks: config.chunks} send := func() (*NativeResult, error) { return call.send(ctx, client) } - result, err := resendRefused(ctx, c, nativeRefused, send) + result, err := withResends(ctx, c, onRefusal(nativeRefused), send) return result, callError(ctx, err) } @@ -68,6 +68,10 @@ func (n nativeCall) send(ctx context.Context, client pb.StoreServiceClient) (*Na return false, nil } err := execute(ctx, client, request, consume) + if failure := refusedStream(err); failure != nil && result == nil { + refused := &NativeResult{Completion: NativeNotStarted, Failure: failure} + return refused, nil + } return result, err } diff --git a/native_test.go b/native_test.go index dbfd712..2ead53b 100644 --- a/native_test.go +++ b/native_test.go @@ -14,6 +14,9 @@ import ( ) func (p *clientTestPeer) nativeExecute(stream grpc.BidiStreamingServer[pb.ExecuteRequest, pb.ExecuteResponse]) error { + if p.mode == "native_refused_stream" && p.received.Load() == 1 { + return refusedStreamError("draining") + } metadata := &Document{ContentType: "application/vnd.example.response", Data: []byte{0xff, 0}} head := &pb.NativeHead{Metadata: metadata, BodyContentType: "application/octet-stream"} if p.mode == "native_invalid_metadata" { @@ -158,6 +161,25 @@ func TestNativeConsumesBytesAndPreservesResponseEvidence(t *testing.T) { } } +func TestNativeResendsARequestWhoseStreamAServerRefused(t *testing.T) { + peer := &clientTestPeer{mode: "native_refused_stream"} + client := clientTestClient(t, peer) + client.resends = newResendBudget() + result, err := client.Native(t.Context(), "records", clientTestNativeRequest()) + if err != nil || result.Err() != nil || peer.received.Load() != 2 { + t.Fatal("refused Native stream was not resent", result, err, peer.received.Load()) + } + client.resends = nil + peer.received.Store(0) + result, err = client.Native(t.Context(), "records", clientTestNativeRequest()) + if err != nil || result.Completion != NativeNotStarted || result.Failure.GetCode() != FailureUnavailable { + t.Fatal("refused Native stream lost its refusal", result, err) + } + if errors.Is(result.Err(), ErrInterrupted) { + t.Fatal("refused Native request marked interrupted", result.Err()) + } +} + func TestNativeSendsBackendRootRequest(t *testing.T) { peer := &clientTestPeer{mode: "native_no_metadata"} client := clientTestClient(t, peer) diff --git a/records.go b/records.go index ed9e8a4..f4f1e87 100644 --- a/records.go +++ b/records.go @@ -45,8 +45,8 @@ func (c *Client) Read(ctx context.Context, store string, options ...ReadOption) // // A single mutation without WithNext, WithMutateConsume or outgoing gRPC metadata // shares long-lived streams with concurrent calls. Any other Mutate uses a gRPC -// stream of its own that follows the Client's servers. A mutation a server refused -// is sent again; nothing else is. +// stream of its own that follows the Client's servers. A mutation a server refused, +// or one that lost to concurrent writers, is sent again; nothing else is. func (c *Client) Mutate(ctx context.Context, store string, options ...MutateOption) ([]*MutationResult, error) { config := applyOptions(options) var completion *pb.MutationOptions @@ -75,13 +75,14 @@ type recordKind[R comparable] struct { result func(*pb.Event) R // refused returns the result of a record a server turned away. refused func(*pb.Failure) R - // isRefused reports a result a server may accept later. - isRefused func(R) bool + // resend returns why the Client sends a result's record again: a server + // may accept it later, or a mutation lost to concurrent writers. + resend func(R) resendCause } -var readKind = recordKind[*ReadResult]{name: "Read", read: true, result: readEventResult, refused: readResultOfRefusal, isRefused: readRefused} +var readKind = recordKind[*ReadResult]{name: "Read", read: true, result: readEventResult, refused: readResultOfRefusal, resend: onRefusal(readRefused)} -var mutateKind = recordKind[*MutationResult]{name: "Mutate", result: mutationEventResult, refused: mutationResultOfRefusal, isRefused: mutationRefused} +var mutateKind = recordKind[*MutationResult]{name: "Mutate", result: mutationEventResult, refused: mutationResultOfRefusal, resend: mutationResend} func readEventResult(event *pb.Event) *ReadResult { wire := event.GetReadResult() @@ -132,7 +133,7 @@ func runRecords[Q any, R comparable](ctx context.Context, c *Client, call record key := laneKey{store: call.store, read: call.kind.read} send := call.sender(c, key) once := func() (R, error) { return send(ctx, last, nil) } - results[0], err = resendRefused(ctx, c, call.kind.isRefused, once) + results[0], err = withResends(ctx, c, call.kind.resend, once) return results, err } produced := 0 @@ -168,7 +169,7 @@ func runRecords[Q any, R comparable](ctx context.Context, c *Client, call record } key := c.lanes.ownKey(call.store, call.kind.read) // Next may block, so only requests it cannot follow wait to share a message. - stream := clientStreamOptions[R]{key: key, known: call.next == nil, next: next, send: call.sender(c, key), refused: call.kind.isRefused, consume: consume} + stream := clientStreamOptions[R]{key: key, known: call.next == nil, next: next, send: call.sender(c, key), resend: call.kind.resend, consume: consume} err = runClientStream(ctx, c, stream) if call.consume == nil { results = growResults(results, produced) @@ -225,7 +226,7 @@ func (call recordCall[Q, R]) sender(c *Client, key laneKey) func(context.Context return call.kind.alone(ctx, c, call.store, command) } if err != nil { - return none, err + return none, callError(ctx, err) } return call.kind.result(event), nil } diff --git a/records_test.go b/records_test.go index 102dbe4..4dceda0 100644 --- a/records_test.go +++ b/records_test.go @@ -172,7 +172,7 @@ func TestMutateBatchSelectsOperationsAndPreservesOrder(t *testing.T) { t.Fatal("wrong business operation", index, mutation) } if index == 1 { - if results[index].GetOutcome() != MutationNotApplied || results[index].GetFailure().GetCode() != FailureConflict { + if results[index].GetOutcome() != MutationNotApplied || results[index].GetFailure().GetCode() != FailurePreconditionFailed { t.Fatal("individual mutation failure lost", results[index]) } } else if results[index].GetOutcome() != MutationApplied { diff --git a/retry.go b/retry.go index dabed69..35743b6 100644 --- a/retry.go +++ b/retry.go @@ -12,9 +12,12 @@ import ( // A Client sends a record again when a server refused it before it began, as // a draining or memory-pressured server or an unavailable backend does: a // mutation NOT_STARTED, a failed read or a Native NOT_STARTED whose Failure is -// UNAVAILABLE. Such a record took no effect, so sending it again cannot apply -// it twice. RESOURCE_EXHAUSTED reports a budget of the request itself, which a -// resend would exceed again. Every other result reaches the caller as it came. +// UNAVAILABLE. It also sends again a mutation NOT_APPLIED with CONFLICT, which +// lost every compare-and-swap or transaction attempt Weir made against +// concurrent writers of its record. Such a record took no effect, so sending +// it again cannot apply it twice. RESOURCE_EXHAUSTED reports a budget of the +// request itself, which a resend would exceed again. Every other result +// reaches the caller as it came. const ( // resendAttempts bounds the sends of one record, the first included. resendAttempts = 4 @@ -28,54 +31,81 @@ const ( resendBurst = 100 ) -// resendBudget holds resend credit; one resend costs resendShare credit and -// every first send earns one. +// resendCause tells why a record may be sent again. Each cause spends a budget +// of its own, so records that keep losing to concurrent writers never use up +// the resends of records a draining server refused. +type resendCause int + +const ( + // noResend marks a result that reaches the caller as it came. + noResend resendCause = iota + // resendRefusal marks a record a server refused before it began. + resendRefusal + // resendConflict marks a mutation that lost to concurrent writers. + resendConflict + resendCauses +) + +// resendBudget holds resend credit for each cause; one resend costs +// resendShare credit of its cause and every first send earns one of each. type resendBudget struct { - credit atomic.Int64 + credit [resendCauses]atomic.Int64 } func newResendBudget() *resendBudget { budget := &resendBudget{} - budget.credit.Store(resendShare * resendBurst) + for cause := range budget.credit { + budget.credit[cause].Store(resendShare * resendBurst) + } return budget } func (b *resendBudget) earn(credits int64) { + for cause := range b.credit { + b.refund(resendCause(cause), credits) + } +} + +// refund adds credit to one cause, up to the burst. +func (b *resendBudget) refund(cause resendCause, credits int64) { + credit := &b.credit[cause] for { - credit := b.credit.Load() - next := min(credit+credits, resendShare*resendBurst) - if next == credit || b.credit.CompareAndSwap(credit, next) { + current := credit.Load() + next := min(current+credits, resendShare*resendBurst) + if next == current || credit.CompareAndSwap(current, next) { return } } } -// spend takes the credit for one resend and reports whether there was enough. -func (b *resendBudget) spend() bool { +// spend takes the credit for one resend of a cause and reports whether there +// was enough. +func (b *resendBudget) spend(cause resendCause) bool { + credit := &b.credit[cause] for { - credit := b.credit.Load() - if credit < resendShare { + current := credit.Load() + if current < resendShare { return false } - if b.credit.CompareAndSwap(credit, credit-resendShare) { + if credit.CompareAndSwap(current, current-resendShare) { return true } } } -// resend reports whether a refused record may be sent as the given attempt, -// after waiting its backoff. It may not once the attempts or the budget run -// out, or when the caller's deadline would not leave the longest wait again -// for the send, which could otherwise turn a refusal into a deadline error -// that may have taken effect. -func (c *Client) resend(ctx context.Context, attempt int) bool { - if c.resends == nil || attempt > resendAttempts || !c.resends.spend() { +// resend reports whether a record may be sent again for a cause as the given +// attempt, after waiting its backoff. It may not once the attempts or the +// cause's budget run out, or when the caller's deadline would not leave the +// longest wait again for the send, which could otherwise turn a result that +// took no effect into a deadline error that may have taken effect. +func (c *Client) resend(ctx context.Context, attempt int, cause resendCause) bool { + if c.resends == nil || attempt > resendAttempts || !c.resends.spend(cause) { return false } longest := resendBackoff << (attempt - 2) wait := longest/2 + rand.N(longest/2+1) if deadline, bounded := ctx.Deadline(); bounded && time.Until(deadline) <= wait+longest { - c.resends.earn(resendShare) + c.resends.refund(cause, resendShare) return false } timer := time.NewTimer(wait) @@ -84,7 +114,7 @@ func (c *Client) resend(ctx context.Context, attempt int) bool { case <-timer.C: return true case <-ctx.Done(): - c.resends.earn(resendShare) + c.resends.refund(cause, resendShare) return false } } @@ -96,16 +126,31 @@ func (c *Client) earn() { } } -// resendRefused sends one record until a server does not refuse it, the +// withResends sends one record until its result calls for no resend, the // attempts or budget run out, or ctx ends, and returns the last outcome. -func resendRefused[T any](ctx context.Context, c *Client, refused func(T) bool, send func() (T, error)) (T, error) { +func withResends[T any](ctx context.Context, c *Client, cause func(T) resendCause, send func() (T, error)) (T, error) { c.earn() for attempt := 2; ; attempt++ { result, err := send() - if err != nil || !refused(result) || !c.resend(ctx, attempt) { + if err != nil { return result, err } + resent := cause(result) + if resent == noResend || !c.resend(ctx, attempt, resent) { + return result, nil + } + } +} + +// onRefusal resends, as refusals, the results refused reports. +func onRefusal[T any](refused func(T) bool) func(T) resendCause { + cause := func(result T) resendCause { + if refused(result) { + return resendRefusal + } + return noResend } + return cause } // refusal reports a Failure with which a server turns away work it may take @@ -118,6 +163,18 @@ func mutationRefused(result *MutationResult) bool { return result.GetOutcome() == pb.MutationOutcome_NOT_STARTED && refusal(result.GetFailure()) } +// mutationResend returns why a mutation that took no effect may apply when sent +// again: a server refused it, or it lost to concurrent writers. +func mutationResend(result *MutationResult) resendCause { + switch { + case mutationRefused(result): + return resendRefusal + case result.GetOutcome() == pb.MutationOutcome_NOT_APPLIED && result.GetFailure().GetCode() == pb.FailureCode_CONFLICT: + return resendConflict + } + return noResend +} + func readRefused(result *ReadResult) bool { return result != nil && refusal(result.Failure) } diff --git a/retry_test.go b/retry_test.go index 9895d95..7ae45c5 100644 --- a/retry_test.go +++ b/retry_test.go @@ -15,19 +15,26 @@ import ( ) // refusingPeer refuses the first refusals[resource] sends of a resource as a -// draining server does, then answers results[resource] or applies the -// mutation and reports a read missing. It counts every send and records the -// documents of applied mutations in order. +// draining server does, then fails conflicts[resource] mutations as lost to +// concurrent writers, then answers results[resource] or applies the mutation +// and reports a read missing. It counts every send and records the documents +// of applied mutations in order. type refusingPeer struct { - mu sync.Mutex - refusals map[string]int - results map[string]*pb.MutationResult - sends map[string]int - applied []string + mu sync.Mutex + refusals map[string]int + conflicts map[string]int + results map[string]*pb.MutationResult + sends map[string]int + applied []string } func newRefusingPeer() *refusingPeer { - peer := &refusingPeer{refusals: make(map[string]int), results: make(map[string]*pb.MutationResult), sends: make(map[string]int)} + peer := &refusingPeer{ + refusals: make(map[string]int), + conflicts: make(map[string]int), + results: make(map[string]*pb.MutationResult), + sends: make(map[string]int), + } return peer } @@ -80,6 +87,10 @@ func (p *refusingPeer) answer(command *pb.Command) *pb.Event { switch { case refused: result = protocol.Mutation(pb.MutationOutcome_NOT_STARTED, failure) + case p.conflicts[resource] > 0: + p.conflicts[resource]-- + conflict := protocol.Fail(pb.FailureCode_CONFLICT, "record changed during every attempt") + result = protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, conflict) case p.results[resource] != nil: result = p.results[resource] default: @@ -157,6 +168,54 @@ func TestClientResendsRecordsAServerRefusedBeforeTheyBegan(t *testing.T) { } } +func TestClientResendsMutationsThatLostToConcurrentWriters(t *testing.T) { + peer := newRefusingPeer() + peer.conflicts["records/s:hot"] = 2 + peer.conflicts["records/s:hotter"] = 10 + client := refusingClient(t, peer) + result, err := refusingPut(t.Context(), client, "records/s:hot") + if err != nil || result.GetOutcome() != MutationApplied || peer.sent("records/s:hot") != 3 { + t.Fatal("conflicted write was not resent until applied", result, err, peer.sent("records/s:hot")) + } + result, err = refusingPut(t.Context(), client, "records/s:hotter") + if err != nil || result.GetOutcome() != MutationNotApplied || result.GetFailure().GetCode() != FailureConflict { + t.Fatal("last conflict lost its evidence", result, err) + } + if peer.sent("records/s:hotter") != resendAttempts { + t.Fatal("unexpected sends", peer.sent("records/s:hotter")) + } + // A call of several records resends a conflicted one before the next + // mutation of its resource. + peer.conflicts["records/s:x"] = 1 + first := putRequest("records/s:x", `{"n":1}`) + later := putRequest("records/s:x", `{"n":2}`) + peer.drain() + results, err := client.Mutate(t.Context(), "records", WithRequests(first, later)) + if err != nil || results[0].GetOutcome() != MutationApplied || results[1].GetOutcome() != MutationApplied { + t.Fatal("conflicted write in a call was not resent", results, err) + } + if applied := peer.drain(); !slices.Equal(applied, []string{`{"n":1}`, `{"n":2}`}) { + t.Fatal("resent write applied out of order", applied) + } +} + +func TestConflictsNeverSpendTheResendsOfRefusals(t *testing.T) { + peer := newRefusingPeer() + client := refusingClient(t, peer) + for client.resends.spend(resendConflict) { + } + peer.conflicts["records/s:hot"] = 1 + result, err := refusingPut(t.Context(), client, "records/s:hot") + if err != nil || result.GetFailure().GetCode() != FailureConflict || peer.sent("records/s:hot") != 1 { + t.Fatal("conflict resent beyond its budget", result, err, peer.sent("records/s:hot")) + } + peer.refuse("records/s:cold", 1) + result, err = refusingPut(t.Context(), client, "records/s:cold") + if err != nil || result.GetOutcome() != MutationApplied || peer.sent("records/s:cold") != 2 { + t.Fatal("refusal not resent after conflicts used their budget", result, err, peer.sent("records/s:cold")) + } +} + func TestClientStopsResendingAfterFourSends(t *testing.T) { peer := newRefusingPeer() peer.refuse("records/s:write", 10) @@ -262,7 +321,7 @@ func TestResendBudgetAllowsATenthOfFirstSendsPlusABurst(t *testing.T) { spent := func(attempts int) int { covered := 0 for range attempts { - if budget.spend() { + if budget.spend(resendRefusal) { covered++ } } @@ -271,6 +330,9 @@ func TestResendBudgetAllowsATenthOfFirstSendsPlusABurst(t *testing.T) { if covered := spent(resendBurst + 50); covered != resendBurst { t.Fatal("burst not bounded", covered) } + if !budget.spend(resendConflict) { + t.Fatal("refusals spent the conflict budget") + } budget.earn(25) if covered := spent(5); covered != 2 { t.Fatal("first sends did not earn a tenth of a resend each", covered) @@ -285,6 +347,7 @@ func TestRefusalsAreThoseBeforeARecordBegan(t *testing.T) { unavailable := protocol.Fail(pb.FailureCode_UNAVAILABLE, "store draining") exhausted := protocol.Fail(pb.FailureCode_RESOURCE_EXHAUSTED, "result exceeds the message size limit") invalid := protocol.Fail(pb.FailureCode_INVALID_ARGUMENT, "invalid record") + conflict := protocol.Fail(pb.FailureCode_CONFLICT, "record changed during every attempt") refusedRead := &ReadResult{Failure: unavailable} exhaustedRead := &ReadResult{Failure: exhausted} invalidRead := &ReadResult{Failure: invalid} @@ -302,6 +365,12 @@ func TestRefusalsAreThoseBeforeARecordBegan(t *testing.T) { {"invalid mutation", mutationRefused(protocol.Mutation(pb.MutationOutcome_NOT_STARTED, invalid)), false}, {"unknown mutation", mutationRefused(protocol.Mutation(pb.MutationOutcome_UNKNOWN, unavailable)), false}, {"missing mutation result", mutationRefused(nil), false}, + {"conflicted mutation", mutationRefused(protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, conflict)), false}, + {"draining mutation resent as a refusal", mutationResend(protocol.Mutation(pb.MutationOutcome_NOT_STARTED, unavailable)) == resendRefusal, true}, + {"conflicted mutation resent as a conflict", mutationResend(protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, conflict)) == resendConflict, true}, + {"unknown conflicted mutation", mutationResend(protocol.Mutation(pb.MutationOutcome_UNKNOWN, conflict)) != noResend, false}, + {"not applied mutation", mutationResend(protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, invalid)) != noResend, false}, + {"missing mutation result resent", mutationResend(nil) != noResend, false}, {"refused read", readRefused(refusedRead), true}, {"exhausted read", readRefused(exhaustedRead), false}, {"invalid read", readRefused(invalidRead), false}, diff --git a/search/index.go b/search/index.go new file mode 100644 index 0000000..175baa7 --- /dev/null +++ b/search/index.go @@ -0,0 +1,188 @@ +package search + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + + weir "github.com/batchstream/weir-go" +) + +// Index holds documents of type T, encoded as JSON, by string ID. Name is an +// index or a single-index alias, which must exist before the first write. +type Index[T any] struct { + Cluster Cluster + Name string + // QueryVisible makes each write return only once later searches see it. + QueryVisible bool +} + +// Hit is one document a search matched. +type Hit[T any] struct { + ID string `json:"_id"` + Score float64 `json:"_score"` + Sort []json.RawMessage `json:"sort"` + Document *T `json:"_source"` +} + +// Get returns the document with the ID, or weir.ErrNotFound. +func (i Index[T]) Get(ctx context.Context, id string) (*T, error) { + documents, err := i.GetMany(ctx, id) + if err != nil { + return nil, err + } + if documents[0] == nil { + return nil, fmt.Errorf("search: %s %s: %w", i.Name, id, weir.ErrNotFound) + } + return documents[0], nil +} + +// GetMany returns the documents with the IDs in one call, in order, with nil +// for each missing one. +func (i Index[T]) GetMany(ctx context.Context, ids ...string) ([]*T, error) { + resources := make([]string, len(ids)) + for n, id := range ids { + resources[n] = i.resource(id) + } + documents, err := i.Cluster.target().Read(ctx, resources) + if err != nil { + return nil, fmt.Errorf("search: %w", err) + } + values := make([]*T, len(documents)) + for n, document := range documents { + if document == nil { + continue + } + value := new(T) + if err := json.Unmarshal(document.GetData(), value); err != nil { + return nil, fmt.Errorf("search: decode %s %s: %w", i.Name, ids[n], err) + } + values[n] = value + } + return values, nil +} + +// Create writes a document that must not exist. An existing one fails it with +// FailurePreconditionFailed. +func (i Index[T]) Create(ctx context.Context, id string, document *T) error { + return i.write(ctx, weir.MutationCreate, id, document) +} + +// Put creates or replaces the document. +func (i Index[T]) Put(ctx context.Context, id string, document *T) error { + return i.write(ctx, weir.MutationPut, id, document) +} + +// Replace replaces a document that must exist. A missing one fails it with +// FailurePreconditionFailed. +func (i Index[T]) Replace(ctx context.Context, id string, document *T) error { + return i.write(ctx, weir.MutationReplace, id, document) +} + +// Delete deletes the document. Deleting a missing document succeeds. +func (i Index[T]) Delete(ctx context.Context, id string) error { + request := &weir.MutateRequest{Resource: i.resource(id), Action: weir.MutationDelete} + return i.mutate(ctx, request) +} + +// Update merges fields, a JSON object, into a document that must exist. A +// []byte or json.RawMessage holds the object's JSON as it is; any other value +// is encoded as JSON. A missing document fails it with +// FailurePreconditionFailed. +func (i Index[T]) Update(ctx context.Context, id string, fields any) error { + if raw, ok := fields.([]byte); ok { + fields = json.RawMessage(raw) + } + expression := struct { + Doc any `json:"doc"` + }{Doc: fields} + data, err := json.Marshal(expression) + if err != nil { + return fmt.Errorf("search: encode update: %w", err) + } + document := &weir.Document{ContentType: updateContentType, Data: data} + request := &weir.MutateRequest{Resource: i.resource(id), Action: weir.MutationAtomicTransform, BackendExpression: document} + return i.mutate(ctx, request) +} + +// Transform runs a Lua program atomically against the document. The program +// returns one function, called as function(current, incoming), where current +// is the stored document or nil and incoming is input, a []byte or +// json.RawMessage holding JSON or another value encoded as JSON, or nil when +// input is nil. Its result replaces the document, and weir.reject(message) +// fails it with FailurePreconditionFailed. See the Lua guide for the rest. +func (i Index[T]) Transform(ctx context.Context, id string, program string, input any) error { + lua := &weir.LuaTransform{Source: []byte(program)} + if input != nil { + data, err := encodeBody(input) + if err != nil { + return fmt.Errorf("search: encode transform input: %w", err) + } + lua.Input = &weir.Document{ContentType: jsonContentType, Data: data} + } + request := &weir.MutateRequest{Resource: i.resource(id), Action: weir.MutationAtomicTransform, Lua: lua} + return i.mutate(ctx, request) +} + +// Search sends a search request body, such as {"query": ..., "size": 10}, to +// the index and returns its hits. A status of 300 or more is a *StatusError. +// Sort holds a hit's sort values exactly, to pass back as search_after. +func (i Index[T]) Search(ctx context.Context, body any) ([]Hit[T], error) { + response, err := i.Do(ctx, http.MethodPost, "/_search", body) + if err != nil { + return nil, err + } + if response.StatusCode >= 300 { + failed := &StatusError{StatusCode: response.StatusCode, Body: response.Body} + return nil, failed + } + var result struct { + Hits struct { + Hits []Hit[T] `json:"hits"` + } `json:"hits"` + } + if err := json.Unmarshal(response.Body, &result); err != nil { + return nil, fmt.Errorf("search: decode %s hits: %w", i.Name, err) + } + return result.Hits.Hits, nil +} + +// Do sends one HTTP request to the index and returns its response, whatever +// its status. The path starts at the index, such as /_search or /_refresh, and +// may carry a query string. Bodies and errors are as for Cluster.Do. +func (i Index[T]) Do(ctx context.Context, method, path string, body any) (*Response, error) { + if i.Name == "" { + return nil, errors.New("search: Index has no Name") + } + request := httpRequest{resource: weir.EncodeSegment(i.Name), method: method, path: path, body: body} + return i.Cluster.send(ctx, request) +} + +func (i Index[T]) write(ctx context.Context, action weir.MutationAction, id string, document *T) error { + if document == nil { + return errors.New("search: document is nil") + } + data, err := json.Marshal(document) + if err != nil { + return fmt.Errorf("search: encode %s %s: %w", i.Name, id, err) + } + encoded := &weir.Document{ContentType: jsonContentType, Data: data} + request := &weir.MutateRequest{Resource: i.resource(id), Action: action, Document: encoded} + return i.mutate(ctx, request) +} + +func (i Index[T]) mutate(ctx context.Context, request *weir.MutateRequest) error { + err := i.Cluster.target().Mutate(ctx, request, i.QueryVisible) + if err != nil { + return fmt.Errorf("search: %s: %w", request.Resource, err) + } + return nil +} + +// resource addresses one document: index/s:id. +func (i Index[T]) resource(id string) string { + resource := weir.EncodeSegment(i.Name) + "/" + weir.EncodeSegment("s:"+id) + return resource +} diff --git a/search/integration_test.go b/search/integration_test.go new file mode 100644 index 0000000..647e6fa --- /dev/null +++ b/search/integration_test.go @@ -0,0 +1,137 @@ +//go:build integration + +package search_test + +import ( + "context" + "errors" + "fmt" + "net/http" + "os" + "strings" + "testing" + "time" + + weir "github.com/batchstream/weir-go" + "github.com/batchstream/weir-go/search" +) + +type item struct { + Group string `json:"group"` + Name string `json:"name"` + Rank int `json:"rank"` +} + +// index returns the pre-created disposable index named by +// WEIR_SEARCH_RESOURCE behind WEIR_SEARCH_STORE. +func index(t *testing.T) search.Index[item] { + t.Helper() + address, store, name := os.Getenv("WEIR_ADDRESS"), os.Getenv("WEIR_SEARCH_STORE"), os.Getenv("WEIR_SEARCH_RESOURCE") + if address == "" || store == "" || name == "" { + t.Skip("requires WEIR_ADDRESS, WEIR_SEARCH_STORE and WEIR_SEARCH_RESOURCE") + } + client, err := weir.New(address) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = client.Close() }) + cluster := search.Cluster{Client: client, Store: store} + items := search.Index[item]{Cluster: cluster, Name: name, QueryVisible: true} + return items +} + +func TestIndexLifecycle(t *testing.T) { + items := index(t) + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + group := fmt.Sprintf("sdk-search-%d", time.Now().UnixNano()) + ids := []string{group + "-a", group + "-b", group + "-c"} + t.Cleanup(func() { + cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + for _, id := range ids { + _ = items.Delete(cleanup, id) + } + }) + + first := &item{Group: group, Name: "first", Rank: 1} + if err := items.Create(ctx, ids[0], first); err != nil { + t.Fatal(err) + } + if err := items.Create(ctx, ids[0], first); weir.CodeOf(err) != weir.FailurePreconditionFailed { + t.Fatal("second create was not refused", err) + } + if err := items.Replace(ctx, ids[1], first); weir.CodeOf(err) != weir.FailurePreconditionFailed { + t.Fatal("replace of a missing document was not refused", err) + } + rank := map[string]int{"rank": 3} + if err := items.Update(ctx, ids[0], rank); err != nil { + t.Fatal(err) + } + const rename = `return function(current, incoming) + if current == nil then return weir.reject("no such item") end + current.name = incoming.name + return current +end` + renamed := map[string]string{"name": "renamed"} + err := items.Transform(ctx, ids[0], rename, renamed) + if err != nil { + t.Fatal(err) + } + err = items.Transform(ctx, ids[1], rename, renamed) + if weir.CodeOf(err) != weir.FailurePreconditionFailed || !strings.Contains(err.Error(), "no such item") { + t.Fatal("rejection lost", err) + } + got, err := items.Get(ctx, ids[0]) + if err != nil || got.Name != "renamed" || got.Rank != 3 || got.Group != group { + t.Fatal("update or transform lost", got, err) + } + second := &item{Group: group, Name: "second", Rank: 2} + third := &item{Group: group, Name: "third", Rank: 1} + if err := items.Put(ctx, ids[1], second); err != nil { + t.Fatal(err) + } + if err := items.Put(ctx, ids[2], third); err != nil { + t.Fatal(err) + } + many, err := items.GetMany(ctx, ids[2], group+"-none") + if err != nil || many[0].Name != "third" || many[1] != nil { + t.Fatal("wrong documents", many, err) + } + + match := map[string]any{"term": map[string]any{"group.keyword": group}} + order := []any{map[string]any{"rank": "desc"}, map[string]any{"name.keyword": "asc"}} + query := map[string]any{"query": match, "sort": order, "size": 2} + hits, err := items.Search(ctx, query) + if err != nil || len(hits) != 2 || hits[0].ID != ids[0] || hits[1].Document.Name != "second" { + t.Fatal("wrong hits", hits, err) + } + query["search_after"] = hits[1].Sort + rest, err := items.Search(ctx, query) + if err != nil || len(rest) != 1 || rest[0].ID != ids[2] { + t.Fatal("search_after did not continue", rest, err) + } + count := map[string]any{"query": match} + response, err := items.Do(ctx, http.MethodPost, "/_count", count) + if err != nil || response.StatusCode != 200 || !strings.Contains(string(response.Body), `"count":3`) { + t.Fatal("wrong count", response, err) + } + bad := map[string]any{"query": map[string]any{"no_such_query": map[string]any{}}} + var failed *search.StatusError + if _, err := items.Search(ctx, bad); !errors.As(err, &failed) || failed.StatusCode != 400 { + t.Fatal("bad query not reported", err) + } + root, err := items.Cluster.Do(ctx, http.MethodGet, "/", nil) + if err != nil || root.StatusCode != 200 { + t.Fatal("cluster root failed", root, err) + } + + for _, id := range ids { + if err := items.Delete(ctx, id); err != nil { + t.Fatal(err) + } + } + if _, err := items.Get(ctx, ids[0]); !errors.Is(err, weir.ErrNotFound) { + t.Fatal("deleted document still found", err) + } +} diff --git a/search/search.go b/search/search.go new file mode 100644 index 0000000..fa27165 --- /dev/null +++ b/search/search.go @@ -0,0 +1,140 @@ +// Package search reads and writes Elasticsearch and OpenSearch documents +// through a Weir Store whose backend is search. +// +// An Index reads and writes documents by ID and runs searches. A Cluster +// sends any other HTTP request: +// +// cluster := search.Cluster{Client: client, Store: "search"} +// products := search.Index[Product]{Cluster: cluster, Name: "products"} +// +// lamp := &Product{Name: "Lamp"} +// err := products.Put(ctx, "p1", lamp) +// product, err := products.Get(ctx, "p1") +// +// match := map[string]any{"match": map[string]any{"name": "lamp"}} +// query := map[string]any{"query": match, "size": 10} +// hits, err := products.Search(ctx, query) +// +// health, err := cluster.Do(ctx, http.MethodGet, "/_cluster/health", nil) +// +// Set a deadline on every context. A failed record is a *weir.Error: check it +// with weir.CodeOf, and with errors.Is against weir.ErrNotFound and +// weir.ErrInterrupted. The package keeps no state of its own; Cluster and +// Index are plain values that are safe to copy and share. +package search + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + + weir "github.com/batchstream/weir-go" + "github.com/batchstream/weir-go/internal/record" +) + +const ( + jsonContentType = "application/json" + updateContentType = "application/vnd.weir.search-update.v1+json" +) + +// Cluster is an Elasticsearch or OpenSearch cluster behind a Weir Store. +type Cluster struct { + Client *weir.Client + // Store names a Weir Store whose backend is search. + Store string +} + +// Response is a complete HTTP response. +type Response struct { + StatusCode int + Header http.Header + Body []byte +} + +// StatusError is a response whose status is 300 or more, from a method that +// expects success. +type StatusError struct { + StatusCode int + Body []byte +} + +func (e *StatusError) Error() string { + return fmt.Sprintf("search: status %d: %s", e.StatusCode, e.Body) +} + +// Do sends one HTTP request to the cluster and returns its response, whatever +// its status. The path starts at the cluster root, such as /_cluster/health, +// and may carry a query string. A []byte or json.RawMessage body is sent as it +// is, any other non-nil body as JSON. A request is sent again only when a +// server refused it before it began. One a lost connection or an unavailable +// backend cut short reports weir.ErrInterrupted and may have run; send it +// again only when that is safe, as for a read. +func (c Cluster) Do(ctx context.Context, method, path string, body any) (*Response, error) { + request := httpRequest{method: method, path: path, body: body} + return c.send(ctx, request) +} + +// httpRequest is one request of Do, relative to resource. +type httpRequest struct { + resource string + method string + path string + body any +} + +func (c Cluster) send(ctx context.Context, request httpRequest) (*Response, error) { + data, err := encodeBody(request.body) + if err != nil { + return nil, fmt.Errorf("search: encode body: %w", err) + } + path := request.path + if path == "" { + path = "/" + } + var body io.Reader + if data != nil { + body = bytes.NewReader(data) + } + message, err := http.NewRequestWithContext(ctx, request.method, path, body) + if err != nil { + return nil, fmt.Errorf("search: %w", err) + } + if data != nil { + message.Header.Set("Content-Type", jsonContentType) + } + native, err := weir.NewHTTPNativeRequest(request.resource, message) + if err != nil { + return nil, fmt.Errorf("search: %w", err) + } + result, err := c.target().Native(ctx, native) + if err != nil { + return nil, fmt.Errorf("search: %s %s: %w", request.method, path, err) + } + head, err := weir.ParseHTTPNativeResponse(result.Response) + if err != nil { + return nil, fmt.Errorf("search: %w", err) + } + response := &Response{StatusCode: head.StatusCode, Header: head.Headers, Body: result.Body} + return response, nil +} + +func (c Cluster) target() record.Target { + target := record.Target{Client: c.Client, Store: c.Store} + return target +} + +// encodeBody returns the bytes of a request body, nil for none. +func encodeBody(body any) ([]byte, error) { + switch value := body.(type) { + case nil: + return nil, nil + case []byte: + return value, nil + case json.RawMessage: + return value, nil + } + return json.Marshal(body) +} diff --git a/search/search_test.go b/search/search_test.go new file mode 100644 index 0000000..d6c83c9 --- /dev/null +++ b/search/search_test.go @@ -0,0 +1,224 @@ +package search + +import ( + "bufio" + "bytes" + "encoding/json" + "errors" + "io" + "net/http" + "testing" + + weir "github.com/batchstream/weir-go" + "github.com/batchstream/weir-go/internal/storetest" + pb "github.com/batchstream/weir-protocol/api/weir/v1" +) + +type product struct { + Name string `json:"name"` + Price int `json:"price"` +} + +// recorder answers commands with a handler and keeps every command it saw. +type recorder struct { + t *testing.T + commands []*pb.Command + answer func(command *pb.Command) []*pb.Event +} + +func (r *recorder) cluster() Cluster { + handler := func(_ string, command *pb.Command) []*pb.Event { + r.commands = append(r.commands, command) + return r.answer(command) + } + client := storetest.Client(r.t, handler) + cluster := Cluster{Client: client, Store: "search"} + return cluster +} + +func (r *recorder) products() Index[product] { + products := Index[product]{Cluster: r.cluster(), Name: "products"} + return products +} + +// httpResponse is the events of a complete HTTP response. +func httpResponse(status string, body string) []*pb.Event { + head := "HTTP/1.1 " + status + "\r\nContent-Type: application/json\r\n\r\n" + metadata := &pb.Document{ContentType: "application/http", Data: []byte(head)} + return storetest.Native(metadata, []byte(body)) +} + +// sentHTTP parses the HTTP request of a Native command. +func sentHTTP(t *testing.T, command *pb.Command) (*http.Request, string) { + t.Helper() + native := command.GetNative() + reader := bufio.NewReader(bytes.NewReader(native.GetRequest().GetData())) + request, err := http.ReadRequest(reader) + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(request.Body) + if err != nil { + t.Fatal(err) + } + return request, string(body) +} + +func TestDoSendsAnHTTPRequestAndReturnsAnyStatus(t *testing.T) { + r := &recorder{t: t, answer: func(*pb.Command) []*pb.Event { return httpResponse("404 Not Found", `{"found":false}`) }} + products := r.products() + query := map[string]any{"size": 0} + response, err := products.Do(t.Context(), http.MethodPost, "/_search?timeout=1s", query) + if err != nil { + t.Fatal(err) + } + if response.StatusCode != 404 || string(response.Body) != `{"found":false}` || response.Header.Get("Content-Type") != "application/json" { + t.Fatal("wrong response", response) + } + request, body := sentHTTP(t, r.commands[0]) + if r.commands[0].GetNative().GetResource() != "products" || request.Method != http.MethodPost || request.RequestURI != "/_search?timeout=1s" { + t.Fatal("wrong request", r.commands[0].GetNative().GetResource(), request.Method, request.RequestURI) + } + if body != `{"size":0}` || request.Header.Get("Content-Type") != "application/json" { + t.Fatal("wrong body", body, request.Header) + } + + raw := []byte("{\"index\":{}}\n{\"name\":\"a\"}\n") + _, err = products.Cluster.Do(t.Context(), http.MethodPost, "/_bulk", raw) + if err != nil { + t.Fatal(err) + } + request, body = sentHTTP(t, r.commands[1]) + if r.commands[1].GetNative().GetResource() != "" || request.RequestURI != "/_bulk" || body != string(raw) { + t.Fatal("raw body or root path changed", request.RequestURI, body) + } + _, err = products.Cluster.Do(t.Context(), http.MethodGet, "", nil) + if err != nil { + t.Fatal(err) + } + request, body = sentHTTP(t, r.commands[2]) + if request.RequestURI != "/" || body != "" || request.Header.Get("Content-Type") != "" { + t.Fatal("wrong root request", request.RequestURI, body) + } +} + +func TestDoInterruptedByAnUnavailableBackend(t *testing.T) { + r := &recorder{t: t, answer: func(*pb.Command) []*pb.Event { return storetest.NativeIncomplete(nil, nil, pb.FailureCode_UNAVAILABLE) }} + _, err := r.cluster().Do(t.Context(), http.MethodGet, "/", nil) + if !errors.Is(err, weir.ErrInterrupted) { + t.Fatal("unavailable backend did not interrupt", err) + } +} + +func TestRecordsUseJSONAndStringIDs(t *testing.T) { + stored := `{"name":"lamp","price":30}` + r := &recorder{t: t} + r.answer = func(command *pb.Command) []*pb.Event { + if read := command.GetRead(); read != nil { + if read.GetResource() == "products/s:p%2F1" { + document := &pb.Document{ContentType: jsonContentType, Data: []byte(stored)} + return storetest.Read(document) + } + return storetest.Missing() + } + return storetest.Applied() + } + products := r.products() + lamp := &product{Name: "lamp", Price: 30} + if err := products.Put(t.Context(), "p/1", lamp); err != nil { + t.Fatal(err) + } + put := r.commands[0].GetMutate() + if put.GetResource() != "products/s:p%2F1" || string(put.GetPut().GetData()) != stored || put.GetPut().GetContentType() != jsonContentType { + t.Fatal("wrong put", put) + } + got, err := products.Get(t.Context(), "p/1") + if err != nil || *got != *lamp { + t.Fatal("wrong document", got, err) + } + _, err = products.Get(t.Context(), "p/2") + if !errors.Is(err, weir.ErrNotFound) { + t.Fatal("missing document not reported", err) + } + many, err := products.GetMany(t.Context(), "p/2", "p/1") + if err != nil || many[0] != nil || *many[1] != *lamp { + t.Fatal("wrong documents", many, err) + } + + products.QueryVisible = true + price := map[string]int{"price": 25} + if err := products.Update(t.Context(), "p/1", price); err != nil { + t.Fatal(err) + } + update := r.commands[len(r.commands)-1].GetMutate() + expression := update.GetAtomicTransform().GetBackendExpression() + if string(expression.GetData()) != `{"doc":{"price":25}}` || expression.GetContentType() != updateContentType { + t.Fatal("wrong update", expression) + } + if update.GetOptions().GetVisibility() != pb.MutationVisibility_QUERY_VISIBLE { + t.Fatal("update did not wait for visibility") + } + if err := products.Update(t.Context(), "p/1", []byte(`{"price":26}`)); err != nil { + t.Fatal(err) + } + raw := r.commands[len(r.commands)-1].GetMutate().GetAtomicTransform().GetBackendExpression() + if string(raw.GetData()) != `{"doc":{"price":26}}` { + t.Fatal("JSON bytes were not sent as they are", string(raw.GetData())) + } + const program = "return function(current, incoming) return incoming end" + err = products.Transform(t.Context(), "p/1", program, lamp) + if err != nil { + t.Fatal(err) + } + lua := r.commands[len(r.commands)-1].GetMutate().GetAtomicTransform().GetLua() + if string(lua.GetSource()) != program || string(lua.GetInput().GetData()) != stored { + t.Fatal("wrong transform", lua) + } +} + +func TestFailedWritesReportTheirCode(t *testing.T) { + r := &recorder{t: t, answer: func(*pb.Command) []*pb.Event { + return storetest.Mutation(pb.MutationOutcome_NOT_APPLIED, pb.FailureCode_PRECONDITION_FAILED) + }} + products := r.products() + lamp := &product{Name: "lamp"} + err := products.Create(t.Context(), "p1", lamp) + if weir.CodeOf(err) != weir.FailurePreconditionFailed { + t.Fatal("wrong error", err) + } + if err := products.Replace(t.Context(), "p1", nil); err == nil { + t.Fatal("nil document accepted") + } +} + +func TestSearchDecodesHitsAndKeepsSortValuesExact(t *testing.T) { + hits := `{"hits":{"total":{"value":2},"hits":[ + {"_id":"p1","_score":null,"sort":[9007199254740993,"lamp"],"_source":{"name":"lamp","price":30}}, + {"_id":"p2","_score":1.5,"_source":{"name":"desk","price":90}}]}}` + status := "200 OK" + r := &recorder{t: t} + r.answer = func(*pb.Command) []*pb.Event { return httpResponse(status, hits) } + products := r.products() + query := map[string]any{"query": map[string]any{"match_all": map[string]any{}}} + found, err := products.Search(t.Context(), query) + if err != nil { + t.Fatal(err) + } + if len(found) != 2 || found[0].ID != "p1" || found[0].Document.Name != "lamp" || found[1].Score != 1.5 { + t.Fatal("wrong hits", found) + } + if string(found[0].Sort[0]) != "9007199254740993" { + t.Fatal("sort value lost precision", string(found[0].Sort[0])) + } + request, body := sentHTTP(t, r.commands[0]) + if request.RequestURI != "/_search" || !json.Valid([]byte(body)) { + t.Fatal("wrong search request", request.RequestURI, body) + } + status = "400 Bad Request" + hits = `{"error":"bad query"}` + _, err = products.Search(t.Context(), query) + var failed *StatusError + if !errors.As(err, &failed) || failed.StatusCode != 400 || string(failed.Body) != hits { + t.Fatal("bad status not reported", err) + } +}