batchwatch

package module
v0.2.1 Latest Latest
Warning

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

 Go to latest Published: Aug 30, 2026 License: MIT Imports: 15 Imported by: 0

README

batchwatch — Go client

Client for batchwatch.dev: crowdsourced measurement of queue time on LLM batch APIs.

Batch endpoints cost 50% of the synchronous ones, but "completes within 24 hours" is impossible to plan around. batchwatch measures what the queue actually does and answers one question: should I use batch for this job?

No dependencies. The standard library — net/http, encoding/json, os, sync, time, context — and nothing else.

Install

Add the module to your go.mod straight from the repo:

go get github.com/batchwatch/client/go@main

or vendor the four .go files — they have no third-party imports. A tagged release for a stable import path is on the way.

import batchwatch "github.com/batchwatch/client/go"

Two lines

bw := batchwatch.New(batchwatch.Options{Token: "bw_..."}) // falls back to $BATCHWATCH_TOKEN
ctx := context.Background()

// 1. before you submit — does this belong in the queue?
if bw.ShouldBatch(ctx, "gpt-5.6-sol", batchwatch.Advice{MaxWait: "15m"}) {
    job := createBatchJob()
    _ = job
} else {
    runSynchronously()
}

// 2. measure it, so the next person gets a better answer
t := bw.Track("gpt-5.6-sol", batchwatch.TrackOpts{InputTokens: batchwatch.Int(9720)})
result := waitFor(job)
t.Done(batchwatch.DoneOpts{OutputTokens: batchwatch.Int(result.OutputTokens)})

Measure closes the measurement for you, including on the error path:

err := bw.Measure("gpt-5.6-sol", batchwatch.TrackOpts{InputTokens: batchwatch.Int(9720)},
    func(t *batchwatch.Tracking) error {
        r, err := waitFor(job)
        if err != nil {
            return err // recorded as failed, then returned untouched
        }
        t.Done(batchwatch.DoneOpts{OutputTokens: batchwatch.Int(r.OutputTokens)})
        return nil
    })

Call bw.Flush() before the process exits to wait for outstanding background submissions.

Get a key with no email and no card:

curl -X POST https://batchwatch.dev/v1/keys -d '{"label":"my pipeline"}'

It fails open, always

If batchwatch is down, slow, or broken, your job must not notice. That is the first requirement, ahead of collecting any data at all.

  • Track() and Done() do no network I/O on your goroutine — submissions run on background goroutines.
  • Two-second timeout by default (per-call context.WithTimeout, BATCHWATCH_TIMEOUT in seconds).
  • Every batchwatch error is swallowed and passed to the optional OnError callback. Nothing is logged unless you wire it up.
  • ShouldBatch() is the one call that blocks, because you want the answer. If it cannot answer you get your own Fallback back — never a guess. The default is false, "run it synchronously": being wrong that way costs money, being wrong the other way blows a deadline.
  • An error returned from your own callback in Measure() is recorded as failed and returned untouched. We swallow our errors, never yours.

fail_open_test.go proves it against a port nothing listens on and against a socket that accepts but never answers.

It never sends your content

No prompts, no completions, no system prompts, no tool calls, no file names. The request body is built from a fixed allowlist — provider, model, mode, endpoint, request count, token counts, timestamps, status — and everything else is dropped by Scrub() on the way out. There is no field to put text in.

no_content_test.go asserts it on the bytes a real HTTP server received, with a positive control so the test cannot pass by the client simply sending nothing.

OutputTokens defaults to nil, never 0

You know your input tokens. You cannot know your output tokens before the model has answered. So the default is absence, not zero — modelled as a *int that is nil until you set it with batchwatch.Int(n).

Zero is not a harmless placeholder: output costs five to six times as much as input, so a saving computed on zero output is systematically too low — measured at 3.4x too low on a real model — and nothing in the response would tell you. If you know a ceiling, pass MaxTokens and the answer comes back labelled as a ceiling.

DoneOpts{OutputTokens: batchwatch.Int(0)} really does send 0: zero is a measurement, absence is not.

Spooling

When a measurement cannot be delivered, the completed record is appended to a JSONL file and replayed later through POST /v1/calls/complete. Losing measurements exactly when the network is bad means losing them exactly when they are most interesting.

  • Default path: $BATCHWATCH_SPOOL, or batchwatch-spool.jsonl in the temp directory (os.TempDir(), resolved with filepath.Join so it is correct on Windows too). Set SpoolDisabled: true to turn it off.
  • Replayed automatically, at most once a minute, right after a successful call — that is the moment we know the network is up. Call bw.FlushSpool(ctx, 0) yourself on shutdown if you want it drained on exit.
  • Spooling requires a token. /v1/calls/complete takes your own timestamps, so it is closed to anonymous callers; without a key a spool file could never be sent, and writing one would just leak disk. bw.Spool is nil when no token is set.
  • The file is capped at 5 MB. Beyond that, measurements are dropped rather than filling your disk.
  • A replayed measurement can arrive twice if the original PATCH reached the server but the response did not. Deliberate: a duplicate is visible in the dataset, a lost measurement is not.
  • The file format is identical across the Python, TypeScript and Go clients, so a spool written by one can be flushed by another.

Configuration

Option Environment Default
Token BATCHWATCH_TOKEN none (anonymous)
BaseURL BATCHWATCH_URL https://batchwatch.dev
Timeout BATCHWATCH_TIMEOUT (seconds) 2s
Spool BATCHWATCH_SPOOL <tempdir>/batchwatch-spool.jsonl
SpoolDisabled — false
Enabled — true
OnError — no-op
HTTPClient — &http.Client{}

Tests

go vet ./...
go test ./...          # or -race, the client is goroutine-heavy

25 tests, no network beyond loopback. They start real HTTP servers on ephemeral ports (httptest.Server) rather than stubbing the transport: the thing under test is network behaviour, so the network should be in the test. The allowlist and fail-open tests carry positive controls, so a client that sent nothing at all would fail them rather than pass.

Licence

MIT

Documentation

Overview

batchwatch client for Go.

Two rules shape this file.

It never fails upwards. If batchwatch is down, slow or broken, the caller's batch job must not notice. Every submission happens on a background goroutine with a short timeout, and every error is swallowed and handed to an optional debug callback. The only error that ever leaves this package is the caller's own.

It never sends content. Prompts, responses, system prompts, tool calls, file names - none of it. The payload is built from a fixed allowlist of fields: provider, model, mode, endpoint, request count, token counts and timestamps. There is no field to put text in, by construction. 

bw := batchwatch.New(batchwatch.Options{Token: "bw_..."})

// 1) before you submit: does this belong in the queue?
if bw.ShouldBatch(ctx, "gpt-5.6-sol", batchwatch.Advice{MaxWait: "15m"}) {
    job := client.Batches.Create(...)
}

// 2) measure it
t := bw.Track("gpt-5.6-sol", batchwatch.TrackOpts{InputTokens: batchwatch.Int(9720)})
// ... wait for the job ...
t.Done(batchwatch.DoneOpts{OutputTokens: batchwatch.Int(4519)})

Standard library only - no dependencies.

Package batchwatch is the Go client for batchwatch.dev.

This file is the on-disk spool for measurements that could not be delivered.

A measurement is worth the most precisely when the network is misbehaving, so that is the worst possible moment to drop one. Undeliverable *completed* measurements are appended to a JSONL file and replayed later through POST /v1/calls/complete.

One JSON object per line, in exactly the shape /v1/calls/complete accepts - the same format the Python, TypeScript and .NET clients use, so a file written by one can be flushed by another.

Replay requires an API key: /v1/calls/complete takes the caller's own timestamps and is closed to anonymous callers for that reason. A client without a token therefore does not spool at all - a file nobody can ever send is just a disk leak (see the Batchwatch constructor).

Threads (goroutines) are handled by a mutex; separate processes sharing one spool file are not. Give each process its own BATCHWATCH_SPOOL if that matters.

Index

Constants

View Source 
const MaxBatch = 500

MaxBatch is the server's ceiling on one POST /v1/calls/complete.

View Source 
const MaxBytes int64 = 5 * 1024 * 1024

MaxBytes caps the spool file. If batchwatch is down for a week, a busy pipeline must not fill the user's disk. Past the cap the measurement is dropped - his machine is not our storage.

View Source 
const Version = "0.2.1"

Version of the client, sent in the user-agent.

Variables

View Source 
var AllowedFields = map[string]bool{
	"provider": true, "model": true, "mode": true, "endpoint": true,
	"requests": true, "input_tokens": true, "output_tokens": true,
	"started_at": true, "ended_at": true, "status": true,
	"ttfb_ms": true, "source": true,
	"acted_verdict": true, "deadline_s": true,
	"quoted_p50_s": true, "quoted_p90_s": true,
}

AllowedFields are the only keys that may leave the machine. Everything else does not exist in the body. The test no_content_test.go pins this list.

This is the Go twin of Scrub; the same seven-plus-source allowlist as the Python and TypeScript clients.

The last four (acted_verdict, deadline_s, quoted_p50_s, quoted_p90_s) are the verdict-accuracy measurement (#101): the advice we ourselves gave, attached to the later completion so the SERVER can compare its measured duration to what we quoted. All numbers and a decision we made - no new PII.

Functions

func Int 

func Int(i int) *int

Int returns a pointer to i. Use it for token counts: a nil *int means "unknown" and is sent as null; Int(0) sends 0, because zero is a measurement and absence is not.

func Scrub added in v0.2.0 

func Scrub(body map[string]any) map[string]any

Scrub drops everything that is not on the allowlist.

This is the last stop before the network. Even though no public method accepts free text, THIS function is the place to point at when someone asks how we know a prompt cannot escape.

Types

type Advice 

type Advice struct {
	MaxWait  string
	Provider string
	Risk     string
	// InputTokens, OutputTokens and MaxTokens are omitted when nil. Output
	// tokens are almost always UNKNOWN before the model answers - leave it
	// nil. Sending 0 would make the server price the saving on zero output,
	// which is systematically too low.
	InputTokens  *int
	OutputTokens *int
	MaxTokens    *int
	// Fallback is returned by ShouldBatch when we cannot answer. Default
	// false = "run it synchronously".
	Fallback bool
}

Advice is the query for ShouldBatch / Advice. Zero-valued fields are omitted from the request; use Int() for the token pointers so that "unknown" (nil) is distinguishable from an explicit value.

type Batchwatch 

type Batchwatch struct {
	Token   string
	Base    string
	Timeout time.Duration
	Enabled bool
	Spool   *Spool
	// contains filtered or unexported fields
}

Batchwatch is the client. Every submission is non-blocking and fails open.

func New 

func New(opts Options) *Batchwatch

New builds a client. Enabled defaults to true.

func (*Batchwatch) AdviceResult 

func (bw *Batchwatch) AdviceResult(ctx context.Context, model string, a Advice) map[string]any

AdviceResult asks the whole verdict. Returns nil if we cannot answer.

OutputTokens is usually UNKNOWN at this point - the model decides them. It is left out unless you pass it. Sending zero would make the server compute the saving on zero output, and output costs five to six times as much as input: the answer would be systematically too low, with nobody able to see it. If you know a ceiling, pass MaxTokens.

func (*Batchwatch) Flush 

func (bw *Batchwatch) Flush()

Flush waits for outstanding submissions. Call it before the process exits.

func (*Batchwatch) FlushSpool 

func (bw *Batchwatch) FlushSpool(ctx context.Context, timeout time.Duration) int

FlushSpool sends everything waiting on disk. Returns the number accepted.

Never returns an error. Records the server rejects as invalid are dropped - they will never become valid.

func (*Batchwatch) Measure 

func (bw *Batchwatch) Measure(model string, opts TrackOpts, fn func(*Tracking) error) error

Measure runs fn and closes the measurement for you, as failed if fn returns an error - and the error is returned untouched.

func (*Batchwatch) ShouldBatch 

func (bw *Batchwatch) ShouldBatch(ctx context.Context, model string, a Advice) bool

ShouldBatch answers true/false. On any doubt you get a.Fallback back.

We never guess on the caller's behalf. The default (Fallback=false) is "run it synchronously", which is the safe way to be wrong: a synchronous call merely costs more, while an unexpected eight-hour queue can take down a product.

func (*Batchwatch) Track 

func (bw *Batchwatch) Track(model string, opts TrackOpts) *Tracking

Track measures one call. It returns immediately - the submission is in flight on a background goroutine. Close it with Done().

func (*Batchwatch) WaitNow 

func (bw *Batchwatch) WaitNow(ctx context.Context, model string, provider, mode string) map[string]any

WaitNow reports what the queue is doing right now, or nil if we cannot say.

type DoneOpts 

type DoneOpts struct {
	// OutputTokens stays null when nil. It is never defaulted to zero: zero
	// is a measurement, absence is not, and the server prices them
	// differently on purpose. Use Int(0) to send an explicit zero.
	OutputTokens *int
	// Status defaults to "completed". Others: failed, expired, cancelled,
	// abandoned.
	Status string
	// TTFBMs is sent only when non-nil.
	TTFBMs *int
}

DoneOpts close a measurement. OutputTokens is a *int: leave it nil when you do not know it and it is sent as null, never 0.

type HTTPError 

type HTTPError struct {
	Status int
	Body   string
}

HTTPError is a non-2xx response. It never reaches the caller of a public method - only the debug OnError callback.

func (*HTTPError) Error 

func (e *HTTPError) Error() string

type Options 

type Options struct {
	// Token is the API key. Falls back to $BATCHWATCH_TOKEN. Optional for
	// measuring, required for replaying a spool.
	Token string
	// BaseURL defaults to $BATCHWATCH_URL or https://batchwatch.dev.
	BaseURL string
	// Timeout is the per-HTTP-call deadline. Defaults to $BATCHWATCH_TIMEOUT
	// seconds or 2s.
	Timeout time.Duration
	// Enabled=false turns every network call into a no-op. The zero value of
	// Options leaves this false, so use New(), which defaults it to true;
	// set DisableEnabled to force it off explicitly.
	Enabled *bool
	// Spool is the spool file path. Empty string uses the default location
	// ($BATCHWATCH_SPOOL or a temp file); set SpoolDisabled to turn it off.
	Spool string
	// SpoolDisabled turns spooling off regardless of Spool.
	SpoolDisabled bool
	// OnError is called on every swallowed error. Wire it to your logger at
	// debug level; nothing is printed otherwise.
	OnError func(error)
	// HTTPClient lets you inject a custom client (proxies, transport). The
	// per-call timeout is applied via context regardless.
	HTTPClient *http.Client
}

Options configure a Batchwatch client. The zero value is usable; every field falls back to an environment variable or a sensible default.

type Record 

type Record = map[string]any

Record is one measurement, in the shape /v1/calls/complete accepts. Using json.RawMessage-free maps keeps the format byte-identical to the other clients, which write plain JSON objects.

type Spool 

type Spool struct {
	// Path is the spool file. Pending is Path + ".pending".
	Path    string
	Pending string
	// MaxBytes is the file-size ceiling. Defaults to the package MaxBytes.
	MaxBytes int64
	// OnError fires on every swallowed error. Wire it to your logger at
	// debug level; nothing is printed otherwise.
	OnError func(error)
	// contains filtered or unexported fields
}

Spool is an append-only JSONL file of completed measurements awaiting delivery. Every method is safe under concurrent goroutines and none of them ever returns the caller an error that would matter: failing to spool must not be worse than the network failure that caused it.

func NewSpool 

func NewSpool(path string) *Spool

NewSpool makes a spool backed by path, with the default size cap.

func (*Spool) Append 

func (s *Spool) Append(record Record) bool

Append stores one completed measurement. Returns true if it was stored.

Never returns an error to the caller: failing to spool must not be worse than the network failure that caused the spooling.

func (*Spool) Keep 

func (s *Spool) Keep(remaining []Record)

Keep puts records back after a partial or failed flush.

func (*Spool) Size 

func (s *Spool) Size() int

Size is the number of records waiting on disk. Best effort, never fails.

func (*Spool) Take 

func (s *Spool) Take() []Record

Take moves everything spooled into the pending file and returns it.

An empty slice means there is nothing to send - including when the spool could not be read at all.

type TrackOpts 

type TrackOpts struct {
	Provider    string
	Mode        string
	Requests    *int
	InputTokens *int
	Endpoint    string
}

TrackOpts configure a measurement. InputTokens is a *int so an unknown count stays null on the wire.

type Tracking 

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

Tracking is the handle for one measurement in flight. Returned by Track().

A Tracking is safe to Done() from any goroutine; Done() is idempotent.

func Track 

func Track(model string, opts TrackOpts) *Tracking

Track is a shortcut using a shared default client, for getting started fast.

func (*Tracking) Done 

func (t *Tracking) Done(opts DoneOpts)

Done closes the measurement. Safe to call more than once; only the first call does anything.

func (*Tracking) Failed 

func (t *Tracking) Failed(status string)

Failed records the job as not completed. An unfinished wait is not a wait.

func (*Tracking) Started 

func (t *Tracking) Started(inputTokens int)

Started updates the token count when it is only known after submission.

Source Files

View all Source files

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.