Add typed MongoDB and search packages so applications stop wrapping the SDK

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
 

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

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 (

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)
 		}

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) {

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
 }
 

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
 

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.

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).

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.