weir

package module
v0.24.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 5, 2026 License: MIT Imports: 31 Imported by: 0

README

Weir Go SDK

CI Go Reference

Typed Go client for Weir.

Overview

Terms
Term Meaning
Store A named backend connection on the Weir server. Every request names a Store.
Resource The path of one record in a Store.
Methods
Method Use
Read(ctx, store, options...) Read records. Results return in input order.
Mutate(ctx, store, options...) Write records. Not a transaction.
Native(ctx, store, request, options...) Send a backend's own command. Use it for queries, scans and counts.
Options
Option Applies to Effect
WithDialOptions New gRPC options for the connection, such as an interceptor or a custom dialer.
WithRequests Read, Mutate The records to send.
WithNext Read, Mutate A callback that produces more records, after those of WithRequests.
WithReadConsume, WithMutateConsume Read, Mutate A callback that receives each result as it arrives, instead of the call returning them.
WithQueryVisible Mutate Wait until normal queries can see the mutations.
WithChunks Native A callback that receives the response body in chunks, instead of Body holding it.
Mutation actions
Action Use
MutationCreate Write a record that must not exist.
MutationReplace Write a record that must exist.
MutationPut Create or replace a record.
MutationDelete Delete a record.
MutationAtomicTransform Change a record with a Lua function or a backend expression.
Backend formats
Backend Resource Document content type
MongoDB database/collection/s:key application/bson. The first field is an _id that matches the key.
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.

Mutation outcomes
Outcome Meaning
MutationApplied The requested semantics were satisfied, including a no-op such as deleting an absent record. It can still carry a Failure from a later acknowledgement.
MutationNotStarted No backend mutation began.
MutationNotApplied The mutation is known not to have applied.
MutationUnknown The mutation may or may not have applied.
Rules
  1. Create one Client and reuse it across goroutines. Close it after callers 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.
  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.
  5. A write may not be visible to the next Search query. Pass WithQueryVisible when it must be.
  6. The application listener uses plaintext gRPC. Restrict its network access.

Requirements

Item Requirement
Go 1.27.3
Weir server Reachable at a host:port address. In Kubernetes, use a headless Service.
Collections and indices Must exist before typed reads and mutations. Native can create them.
Docker For the quick start only

Quick start

This writes one record to MongoDB through Weir and reads it back.

Step 1. Start MongoDB and Weir with steps 1 to 4 of the Weir quick start.

Step 2. In a second terminal, create a module and add the SDK.

mkdir hello
cd hello
go mod init example.com/hello
go get github.com/batchstream/weir-go@v0.24.1 go.mongodb.org/mongo-driver/v2/bson

Step 3. Save this as main.go.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	weir "github.com/batchstream/weir-go"
	"go.mongodb.org/mongo-driver/v2/bson"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	client, err := weir.New("127.0.0.1:7447")
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	const storeName = "mongo"
	const resource = "weir_example/records/s:hello"

	// Write one record. The first field is an _id that matches the resource key.
	record := bson.D{{Key: "_id", Value: "hello"}, {Key: "message", Value: "hi"}}
	data, err := bson.Marshal(record)
	if err != nil {
		log.Fatal(err)
	}
	document := &weir.Document{ContentType: "application/bson", Data: data}
	write := &weir.MutateRequest{Resource: resource, Action: weir.MutationPut, Document: document}
	written, err := client.Mutate(ctx, storeName, weir.WithRequests(write))
	if err != nil {
		log.Fatal(err)
	}
	if written[0].GetFailure() != nil {
		log.Fatal(written[0].GetFailure())
	}
	fmt.Println("outcome:", written[0].GetOutcome())

	// Read the record back.
	read := &weir.ReadRequest{Resource: resource}
	results, err := client.Read(ctx, storeName, weir.WithRequests(read))
	if err != nil {
		log.Fatal(err)
	}
	result := results[0]
	if result.Failure != nil {
		log.Fatal(result.Failure)
	}
	if result.Missing {
		log.Fatal("record not found")
	}

	var stored bson.M
	err = bson.Unmarshal(result.Document.Data, &stored)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("message:", stored["message"])
}

Step 4. Run it.

go run .

Expected output:

outcome: APPLIED
message: hi

Step 5. Stop Weir and MongoDB with step 6 of the Weir quick start.

Documentation

Guide Contents
Connections Addresses, gRPC options, Kubernetes DNS names, keepalive and closing
Operations Reads, writes, visibility, results and outcomes
Lua Lua transforms
Native Backend commands and HTTP helpers
Streaming Callbacks, streams, deadlines and resending refused records
Development Tests, real backend tests and releases
API reference Every exported type and function
Example Shows
read A typed read
basic A mutation, then many reads in one call
native A native command and its response

Development

go test -race ./...
go vet ./...

Default tests need no databases. See Development for real backend tests.

Repository Purpose
weir Server
weir-go Go client SDK (this repository)
weir-protocol Public gRPC schema and validation
weir-charts Helm chart
weir-tests System tests and benchmarks

License

MIT. See LICENSE.

Documentation

Overview

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.

Index

Constants

View Source 
const (
	MutationNotStarted = pb.MutationOutcome_NOT_STARTED
	MutationNotApplied = pb.MutationOutcome_NOT_APPLIED
	MutationApplied    = pb.MutationOutcome_APPLIED
	MutationUnknown    = pb.MutationOutcome_UNKNOWN

	NativeCompletionUnconfirmed = pb.NativeCompletion_NATIVE_COMPLETION_UNSPECIFIED
	NativeNotStarted            = pb.NativeCompletion_NATIVE_NOT_STARTED
	NativeResponseComplete      = pb.NativeCompletion_RESPONSE_COMPLETE
	NativeResponseIncomplete    = pb.NativeCompletion_RESPONSE_INCOMPLETE

	FailureInvalidArgument    = pb.FailureCode_INVALID_ARGUMENT
	FailureUnauthenticated    = pb.FailureCode_UNAUTHENTICATED
	FailurePermissionDenied   = pb.FailureCode_PERMISSION_DENIED
	FailureTargetNotFound     = pb.FailureCode_TARGET_NOT_FOUND
	FailurePreconditionFailed = pb.FailureCode_PRECONDITION_FAILED
	FailureConflict           = pb.FailureCode_CONFLICT
	FailureUnsupported        = pb.FailureCode_UNSUPPORTED
	FailureResourceExhausted  = pb.FailureCode_RESOURCE_EXHAUSTED
	FailureUnavailable        = pb.FailureCode_UNAVAILABLE
	FailureCancelled          = pb.FailureCode_CANCELLED
	FailureDeadlineExceeded   = pb.FailureCode_DEADLINE_EXCEEDED
	FailureInternal           = pb.FailureCode_INTERNAL
)

Variables

View Source 
var ErrClosed = errors.New("Weir client is closed")

ErrClosed reports a call started after Close. Calls already running when Close is called end with the gRPC cancellation error instead.

Functions

func EncodeSegment 

func EncodeSegment(segment string) string

EncodeSegment encodes one decoded resource path segment, including slashes.

Types

type Client 

type Client struct {
	// contains filtered or unexported fields
}

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.

func New 

func New(address string, options ...ClientOption) (*Client, error)

New returns a Client for a Weir application address such as "weir:7447". The address is a host name or IP literal with an explicit port. The Client connects on first use and reconnects after failures. A DNS name that resolves to several servers, such as a headless Service, spreads streams across them. The Client looks the name up every five seconds: a server that joins takes its share of streams, and one that stops resolving gets no new records while its streams finish the ones they carry.

func (*Client) Close 

func (c *Client) Close() error

Close cancels active requests and closes the connection. Concurrent and repeated calls wait for the first and return its result.

func (*Client) Mutate 

func (c *Client) Mutate(ctx context.Context, store string, options ...MutateOption) ([]*MutationResult, error)

Mutate sends the mutations of WithRequests and then of WithNext, and returns one result per mutation in input order. Mutations of one resource run in input order, each after the previous one's result, including after a failure; those of different resources may run concurrently. It is not a transaction. A later error preserves confirmed results, and unconfirmed positions stay nil and may have applied. With WithMutateConsume results go to its callback and Mutate returns nil results. It needs at least one mutation, or returns an error.

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.

func (*Client) Native 

func (c *Client) Native(ctx context.Context, store string, request *NativeRequest, options ...NativeOption) (*NativeResult, error)

Native sends one backend command and returns its response. Backend errors remain native response data, and a validated result remains evidence alongside a later RPC error. The body is collected into NativeResult.Body unless WithChunks consumes it incrementally. The request is sent again only when a server refused it with NativeNotStarted before any response.

func (*Client) Read 

func (c *Client) Read(ctx context.Context, store string, options ...ReadOption) ([]*ReadResult, error)

Read sends the reads of WithRequests and then of WithNext, and returns one result per read in input order. A later error preserves confirmed results, and unconfirmed positions stay nil. With WithReadConsume results go to its callback and Read returns nil results. It needs at least one read, or returns an error.

A single read without WithNext, WithReadConsume or outgoing gRPC metadata shares long-lived streams with concurrent calls. Any other Read uses a gRPC stream of its own that follows the Client's servers. A read a server refused is sent again; nothing else is.

type ClientOption added in v0.24.0 

type ClientOption func(*clientOptions)

ClientOption configures a Client.

func WithDialOptions added in v0.24.0 

func WithDialOptions(options ...grpc.DialOption) ClientOption

WithDialOptions adds gRPC options to a Client's connection, such as a stats handler, an interceptor or a custom dialer. They can replace its defaults for transport credentials and keepalive. They cannot change what a Client relies on: gRPC never resends a record, message sizes stay at the protocol's limit, service configs have no effect, and its own resolver and balancer follow its servers. Stats handlers and interceptors see gRPC streams rather than calls, so they suit stream metrics, not per-call tracing. Several WithDialOptions add their options in order, and a nil one is ignored.

type Document 

type Document = pb.Document

Leaf DTOs preserve document bytes and backend evidence without exposing any transport framing or protobuf operation selection.

type Failure 

type Failure = pb.Failure

type FailureCode 

type FailureCode = pb.FailureCode

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.

type HTTPNativeResponse 

type HTTPNativeResponse struct {
	StatusCode int
	Headers    http.Header
}

HTTPNativeResponse describes an application/http metadata document. Its body stays in NativeResult.Body or goes to the WithChunks callback, and this helper never retains it.

func ParseHTTPNativeResponse 

func ParseHTTPNativeResponse(response *NativeResponse) (*HTTPNativeResponse, error)

ParseHTTPNativeResponse parses a complete HTTP status line and header block from Native response metadata. No body bytes may follow the terminating CRLF. Other native formats stay opaque and do not require this optional helper.

type LuaTransform 

type LuaTransform = pb.LuaTransform

type MutateOption added in v0.24.0 

type MutateOption = Option[*MutateRequest]

MutateOption configures a Mutate.

func WithMutateConsume added in v0.24.0 

func WithMutateConsume(consume func(ctx context.Context, index int, result *MutationResult) error) MutateOption

WithMutateConsume receives each mutation's result as its record completes, instead of the call collecting them, and Mutate returns nil results. The index is the record's position in the input, from 0. Each result is an individual acknowledgement that a later error does not revoke; records without a result may have applied. Calls run one at a time and hold back the input when slow; an error from consume ends the call. It must honor its context, as a call does not return before it does.

func WithQueryVisible added in v0.24.0 

func WithQueryVisible() MutateOption

WithQueryVisible requires every mutation of the call to be visible to later normal queries before an APPLIED result without Failure. APPLIED with Failure confirms application only. Without it, a mutation completes on application acknowledgement, with no separate visibility guarantee.

type MutateRequest 

type MutateRequest struct {
	Resource          string
	Action            MutationAction
	Document          *Document
	Lua               *LuaTransform
	BackendExpression *Document
}

MutateRequest describes one mutation. Create, Put and Replace require Document; Delete accepts no payload; AtomicTransform requires Lua or BackendExpression.

type MutationAction 

type MutationAction uint8

MutationAction selects one business operation without a protobuf oneof. 

const (
	MutationCreate MutationAction = iota + 1
	MutationPut
	MutationReplace
	MutationDelete
	MutationAtomicTransform
)

type MutationOutcome 

type MutationOutcome = pb.MutationOutcome

type MutationResult 

type MutationResult = pb.MutationResult

type NativeCompletion 

type NativeCompletion = pb.NativeCompletion

type NativeOption added in v0.24.0 

type NativeOption func(*nativeOptions)

NativeOption configures a Native call.

func WithChunks added in v0.24.0 

func WithChunks(consume func(ctx context.Context, response *NativeResponse, chunk []byte) error) NativeOption

WithChunks receives a Native response's body incrementally, instead of the call collecting it into NativeResult.Body. Each nonempty chunk finishes before the next one, and consume must honor its context. An empty response invokes no callback. Do not mutate the response metadata, which stays attached to the result.

type NativeRequest 

type NativeRequest = pb.NativeRequest

NativeRequest carries an opaque adapter-owned document. Keep its bytes immutable until Native returns. Empty request data is permitted. Resource is an optional target prefix; empty selects the backend root when supported.

func NewHTTPNativeRequest 

func NewHTTPNativeRequest(resource string, request *http.Request) (*NativeRequest, error)

NewHTTPNativeRequest serializes a standard HTTP/1.1 request using an origin-form request target. It consumes and closes request.Body. Native imposes no method, path or parameter allowlist. Resource is an optional target prefix; empty addresses the backend root. Host does not select a Store.

type NativeResponse 

type NativeResponse = pb.NativeHead

type NativeResult 

type NativeResult struct {
	Response   *NativeResponse
	Body       []byte
	Completion NativeCompletion
	Failure    *Failure
}

NativeResult retains validated response and completion evidence. Completion is NativeCompletionUnconfirmed until a terminal acknowledgement arrives; Response may already be available. An RPC error never revokes a confirmed terminal. Completion describes response transport, not whether backend writes applied. Body holds the response body of a call without WithChunks, as far as it arrived.

type Option added in v0.24.0 

type Option[Q any] func(*callOptions[Q])

Option configures one Read or Mutate call whose requests have type Q.

func WithNext added in v0.24.0 

func WithNext[Q any](next func(ctx context.Context) (Q, error)) Option[Q]

WithNext produces requests one at a time, after those of WithRequests have entered the window of 32 records and 8 MiB, without waiting for their results. Next returns io.EOF after the last request. Each request is validated as it is produced. An invalid request or another error from Next ends the input there: the records before it still finish, and the call then returns that error. Next runs concurrently with a consume callback, so synchronize what they share. It must honor its context, and return promptly once that ends. Keep a produced request and its document bytes unchanged until the call returns, or with a consume callback until its result arrives there.

Without a consume callback, results accumulate until the call returns, and memory grows with the number of records.

func WithRequests added in v0.24.0 

func WithRequests[Q any](requests ...Q) Option[Q]

WithRequests adds requests the call already holds. They are all validated before anything is sent, so an invalid one rejects the whole call and WithNext is never called. Several WithRequests add their requests in order. Keep the requests and their document bytes unchanged until the call returns, and after a call its context ended do not reuse them at all: a single record's shared stream may still be reading them.

type ReadOption added in v0.24.0 

type ReadOption = Option[*ReadRequest]

ReadOption configures a Read.

func WithReadConsume added in v0.24.0 

func WithReadConsume(consume func(ctx context.Context, index int, result *ReadResult) error) ReadOption

WithReadConsume receives each read's result as its record completes, instead of the call collecting them, and Read returns nil results. The index is the record's position in the input, from 0. Results delivered before a later error remain valid. Calls run one at a time and hold back the input when slow; an error from consume ends the call. It must honor its context, as a call does not return before it does.

type ReadRequest 

type ReadRequest = pb.ReadRequest

type ReadResult 

type ReadResult struct {
	Document *Document
	Missing  bool
	Failure  *Failure
}

ReadResult contains exactly one of Document, Missing, or Failure. Missing confirms document absence after a successful read. FailureTargetNotFound reports a missing adapter-owned target, never a missing document.

Source Files

View all Source files

Directories

Path Synopsis
examples
basic command
Basic demonstrates typed finite read and mutation batches addressed to one Store.
 Basic demonstrates typed finite read and mutation batches addressed to one Store.
native command
Native sends one backend request and consumes its response incrementally.
 Native sends one backend request and consumes its response incrementally.
read command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.