attos

package module
v1.4.0 Latest Latest
Warning

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

 Go to latest Published: Oct 31, 2026 License: Apache-2.0 Imports: 39 Imported by: 0

README

attos-go

attos-go is the Go client SDK for Attos, a zero-latency state distribution engine. Attos Cloud compiles your dataset into a static .attos blob. This SDK downloads it, memory-maps it, and serves lookups in-process: lock-free and allocation-free on the read path.

3-minute quickstart

You need Go 1.24+ and a free Attos Cloud account.

1. Sign up at https://attoscloud.com/auth/signup. Every signup gets a Free-tier personal project.

2. Create a dataset. Save this as sample.ndjson:

{"user_id": "user_1", "attos_value": "free"}
{"user_id": "user_2", "attos_value": "pro"}
{"user_id": "user_3", "attos_value": "enterprise"}

In the dashboard open Datasets → Create dataset and upload the file. When the status is ready, copy the dataset ID shown under the dataset name.

3. Create an API key under Settings → API Keys. Copy it now: it is only shown once. New keys start with at_live_; keys created earlier start with nh_live_ and still work.

4. Look up a key.

mkdir attos-demo && cd attos-demo
go mod init attos-demo
go get github.com/Attos-Labs/attos-go@latest
export ATTOS_API_KEY="at_live_..."
export ATTOS_DATASET_ID="<dataset id from step 2>"

// main.go
package main

import (
	"context"
	"fmt"
	"log"
	"os"

	attos "github.com/Attos-Labs/attos-go"
)

func main() {
	s, err := attos.NewSynchronizer(
		context.Background(),
		os.Getenv("ATTOS_DATASET_ID"),
		attos.WithBaseURL("https://api.attoscloud.com"),
		attos.WithAPIKey(os.Getenv("ATTOS_API_KEY")),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer s.Close()

	v, err := s.GetString("user_2")
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("user_2 =", v)
}

go run .
# user_2 = pro

NewSynchronizer downloads the blob before it returns, so the first Get already works. The blob is cached in ./.attos/cache, and the SDK re-checks for a new version every 30 seconds in the background.

If it fails: 401 means the key is wrong, 404 means the dataset ID is wrong, and 409 means the dataset is still compiling, so retry in a few seconds.

Create datasets via the API

You can also create a dataset without the dashboard. Upload a CSV or NDJSON file with your API key:

curl -X POST https://api.attoscloud.com/api/v1/datasets/compile \
  -H "X-API-Key: $ATTOS_API_KEY" \
  -F file=@sample.ndjson \
  -F name=my-dataset

The input must have an attos_value column; every other column is a key dimension. The response is 201 with id (use it as ATTOS_DATASET_ID), name, row_count, blob_size, compression, compression_note and status. name is optional and defaults to the file name without its extension. Instead of a multipart upload you can send the NDJSON as the request body and set the name with an X-Dataset-Name header.

Compression is off by default. To turn on LZ4, add -F compression=lz4 and optionally -F block_size=<records> (for a raw body, use the X-Attos-Compression and X-Attos-Block-Size headers). With LZ4, each lookup decompresses one block, which costs roughly 0.4 µs per record in the block and allocates. There is no setting for the input format or the blob format version. To replace the data of an existing dataset, PUT /api/v1/datasets/{id}/data with the same API key; it counts against your plan's monthly update limit.

Listing and deleting datasets and creating API keys need a dashboard login and are not available with an API key.

Local files

You can also map a .attos file directly, with no cloud involved:

m, err := attos.Open("/var/data/rules.attos")
if err != nil {
	log.Fatal(err)
}
defer m.Close()

val, err := m.GetString("user_123")

Cloud sync options

Pass these to attos.NewSynchronizer(ctx, stashID, opts...):

Option Description
WithBaseURL(string) Control plane URL. Default http://localhost:8080, so always set it for Attos Cloud.
WithAPIKey(string) API key, sent as X-API-Key.
WithSyncInterval(time.Duration) Poll interval. Default 30s; 0 disables background polling.
WithCacheDir(string) Where the downloaded blob is cached. Default ./.attos/cache.
WithNodeID(string) Node ID reported to the cloud. Defaults to ATTOS_NODE_ID, then the hostname. Nodes are counted by distinct ID.
WithCompression(bool) Ask for the gzip transfer copy. Default true.
WithLogger(*log.Logger) Logger for sync and telemetry messages.
WithTelemetryDisabled() Turn the telemetry emitter off.
WithOpenConfig(attos.OpenConfig) Checksum validation and memory pinning for the mapped blob.

Batch lookups and typed values

out := make([][]byte, len(keys))
found, err := m.GetMany(keys, out) // out[i] == nil when keys[i] is absent

GetMany pins the mapping once and reads a group of keys' table entries together, so it costs a little less per key than repeated Get on large files (measured at roughly 6-10 % on the 15M-key benchmark file; ATTOS_BENCH_FILE=your.attos go test -run xxx -bench 'Single|Batch' measures your own). Batch calls count as lookups but are not latency-sampled, and make no allocations.

Values are opaque bytes. If yours are structured, Schema reads fixed layouts (big-endian numbers, u16-length strings) straight from the value, using the same layout as the Rust and Java SDKs:

sc := attos.NewSchema(attos.Field{"id", attos.U64}, attos.Field{"name", attos.String})
name, err := sc.String(value, "name")

Metrics

http.Handle("/metrics", attos.MetricsHandler(sync)) serves Prometheus text: lookups, misses and hit ratio, a sampled lookup-latency histogram (1 in 256 lookups), hot swaps, CRC and sync errors, and attos_file_info{format_version,etag,stash} for the file in use. Scraping reads counters the SDK keeps anyway; lookups stay allocation-free.

Sizing against Redis: attos-size

go run github.com/Attos-Labs/attos-go/cmd/attos-size data.csv          # Attos input format (attos_value column)
go run github.com/Attos-Labs/attos-go/cmd/attos-size -kv export.csv    # headerless key,value CSV, e.g. a Redis export
go run github.com/Attos-Labs/attos-go/cmd/attos-size -redis host:6379  # sample a live Redis (read-only commands only)

It samples up to 200,000 rows (-sample), builds real Attos files from the sample in three variants (Fast, Lean, Compressed), extrapolates the file size, and prints RAM saved against Redis and the equivalent monthly Redis cost. For file input the Redis size is an estimate (key + value + an assumed 56 bytes per key, -redis-overhead); in live mode it is measured with MEMORY USAGE. The price (-redis-usd-per-gib-month, default 25) and -replicas are assumptions you should replace with your own bill. Use -json for machine-readable output. Live mode sends only SCAN, TYPE, GET, MEMORY USAGE and DBSIZE; the password is read from the environment variable named by -redis-password-env.

Signed artifacts

pub, _ := attos.ParsePublicKey("<base64 Ed25519 public key>")
s, err := attos.NewSynchronizer(ctx, stashID, attos.WithSignatureKeys(map[string]ed25519.PublicKey{"2026-10": pub}))

With keys set, the synchronizer installs a file only if its X-Attos-Signature header verifies (Ed25519 over the file's SHA-256, bound to the stash id), and re-verifies the cached copy on restart. A missing or invalid signature is refused (ErrSignatureMissing, ErrSignatureInvalid), the current file keeps serving, and attos_signature_errors_total counts it. The file format is unchanged, and deltas are not requested while signing is enforced because they are unsigned. The signed message is "attos-artifact-v1\0" + stash_id + "\0" + SHA-256(file), and the header is alg=ed25519; kid=<id>; sig=<base64>. The control plane must sign the files for this to work; do not enable enforcement until it does.

Errors

  • attos.ErrKeyNotFound: the key is not in the current dataset.
  • attos.ErrInvalidMagic, attos.ErrCorruptedBinary, attos.ErrChecksumMismatch: the blob failed validation.
  • *attos.StatusError: the cloud returned a non-200 status. Quota() is true for plan limits (402/429); the SDK keeps serving the last good blob.

Format compatibility

This SDK reads .attos format versions 1 and 2 (PTHash, LZ4 blocks, 40-bit offsets, lean keys) from the first release that contains them, and fails closed with an error that names the unsupported feature and says to upgrade the SDK when it meets a newer file. The full table of format features against SDK versions is in docs/FORMAT_COMPATIBILITY.md.

Documentation

Index

Constants

View Source 
const (
	// MaxKeyLen is the largest key (in bytes) a v1 blob can store.
	MaxKeyLen = 65535
	// MaxValueLen is the largest value (in bytes) a v1 blob can store.
	MaxValueLen = 1 << 20
)
View Source 
const (
	DefaultSyncInterval = 30 * time.Second
	QuotaPause          = 5 * time.Minute
	MaxRetryAfter       = time.Hour
)

Sync protocol constants (CONTRACTS.md §4). The default poll interval is the same in the Go, Rust and Java SDKs.

View Source 
const DefaultRetireGrace = 2 * time.Second

DefaultRetireGrace is the default zero-copy validity window after a swap.

View Source 
const (
	DefaultTelemetryInterval = 60 * time.Second
)
View Source 
const SignatureHeader = "X-Attos-Signature"

SignatureHeader is the response header carrying the detached signature.

View Source 
const Version = "1.4.0"

Version is the attos-go SDK version reported in User-Agent and telemetry.

Variables

View Source 
var (
	ErrKeyNotFound     = errors.New("key not found")
	ErrCorruptedBinary = errors.New("corrupted binary")
	ErrInvalidMagic    = errors.New("invalid magic bytes")
	ErrInvalidVersion  = errors.New("invalid version")
	// ErrUnsupportedFeature is returned (via *FeatureError) for a well-formed file that uses a
	// format feature newer than this SDK knows. The fix is to upgrade the SDK.
	ErrUnsupportedFeature = errors.New("unsupported format feature")
	ErrChecksumMismatch   = errors.New("checksum mismatch")
	ErrMapClosed          = errors.New("map closed")
	ErrMlockFailed        = errors.New("mlock failed")
	// ErrNoMap is returned by Synchronizer lookups before the first blob is loaded.
	ErrNoMap = errors.New("no blob loaded")
)

Reader / runtime errors (CONTRACTS.md §1.6).

View Source 
var (
	ErrMissingValueColumn = errors.New("missing attos_value column")
	ErrNoDimensions       = errors.New("no dimension columns")
	ErrBlobTooLarge       = errors.New("blob too large")
	ErrUnplaceable        = errors.New("unplaceable bucket")
	ErrMalformedRow       = errors.New("malformed row")
	ErrDuplicateKey       = errors.New("duplicate key")
	ErrLimit              = errors.New("size limit exceeded")
)

Builder errors (CONTRACTS.md §1.5 / §3). Typed errors unwrap to these.

View Source 
var (
	// ErrSignatureMissing: signing is enforced but the response carried no signature.
	ErrSignatureMissing = errors.New("attos: dataset signature missing")
	// ErrSignatureInvalid: the signature is malformed, names an unknown key or does not verify.
	ErrSignatureInvalid = errors.New("attos: dataset signature invalid")
)
View Source 
var ErrBatchOutTooSmall = errors.New("batch output slice shorter than keys")

ErrBatchOutTooSmall is returned by GetMany when out is shorter than keys.

View Source 
var ErrSchema = errors.New("typed value does not match schema")

ErrSchema is returned for values that do not fit their Schema.

View Source 
var ErrTruncatedDownload = errors.New("attos sync: truncated download")

ErrTruncatedDownload is returned when the body is shorter than Content-Length.

Functions

func MetricsHandler 

func MetricsHandler(src promWriter) http.Handler

MetricsHandler serves the Prometheus exposition of a *Map or *Synchronizer, for example http.Handle("/metrics", attos.MetricsHandler(sync)).

func MmapFile 

func MmapFile(path string) ([]byte, *os.File, error)

MmapFile maps path read-only. Deprecated: retained for API compatibility.

func MunmapFile 

func MunmapFile(data []byte, f *os.File)

MunmapFile releases a mapping returned by MmapFile. Deprecated.

func ParsePublicKey 

func ParsePublicKey(b64 string) (ed25519.PublicKey, error)

ParsePublicKey decodes a base64 (standard or URL-safe, padded or not) Ed25519 public key.

func SignArtifact 

func SignArtifact(priv ed25519.PrivateKey, kid, stashID string, digest []byte) string

SignArtifact signs a file digest the way the control plane does (for tests and tooling).

func ValidateFile 

func ValidateFile(path string) error

ValidateFile fully validates a blob on disk (magic, version, flag, size arithmetic and CRC32) without keeping it mapped. This is the validation the sync path applies before publishing a downloaded blob.

Types

type BuildStats 

type BuildStats struct {
	Rows            uint64 // n
	Slots           uint64 // S
	Buckets         uint64 // B
	MaxDisplacement uint32
	Bytes           int64 // bytes written, including the CRC trailer
	CRC32           uint32
}

BuildStats describes a compiled blob.

func Compile 

func Compile(ctx context.Context, src io.Reader, format Format, dst io.Writer) (BuildStats, error)

Compile reads src fully, writes one complete v1 blob to dst and returns its statistics. It mirrors attos-cloud/engine/compiler.Compile (CONTRACTS.md §3) and produces byte-identical output. Nothing is written to dst if parsing, duplicate detection or placement fails.

type Builder 

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

Builder accumulates (key, value) rows and compiles them into a v1 blob. A Builder is not safe for concurrent use.

func NewBuilder 

func NewBuilder() *Builder

NewBuilder returns an empty Builder.

func NewSpillBuilder 

func NewSpillBuilder(dir string) (*Builder, error)

NewSpillBuilder returns a Builder that stores row data in a temporary file under dir ("" means os.TempDir) and keeps only a small index in memory. Output is byte-for-byte identical to NewBuilder. Call Close when done to remove the temporary file.

func (*Builder) Add 

func (b *Builder) Add(key, value []byte, line int) error

Add appends one row. line is used only for error reporting (use 0 if unknown). Add does not check for duplicates until the next BuildFrom or Compile call.

func (*Builder) BuildFrom 

func (b *Builder) BuildFrom(ctx context.Context, src io.Reader, format Format) error

BuildFrom parses src in the given format and adds its rows. On any error the Builder is left exactly as it was before the call. Line numbers in errors are 1-based physical line numbers within src.

Error precedence matches the reference implementation: a parse error anywhere in src wins; otherwise the earliest row that exceeds a size limit or repeats an earlier key is reported.

func (*Builder) BuildFromCSV 

func (b *Builder) BuildFromCSV(csvPath string) error

BuildFromCSV parses an RFC 4180 CSV file (header row required) and adds its rows.

func (*Builder) BuildFromFile 

func (b *Builder) BuildFromFile(path string) error

BuildFromFile parses a CSV or NDJSON file, detecting the format (§1.5).

func (*Builder) BuildFromNDJSON 

func (b *Builder) BuildFromNDJSON(jsonPath string) error

BuildFromNDJSON parses an NDJSON file and adds its rows.

func (*Builder) Close 

func (b *Builder) Close() error

Close releases the temporary file of a spilling Builder. It is a no-op for NewBuilder builders.

func (*Builder) Compile 

func (b *Builder) Compile(outPath string) error

Compile writes a version 2 (.attos) file without compression. See CompileWithOptions.

func (*Builder) CompileFile 

func (b *Builder) CompileFile(ctx context.Context, outPath string) (BuildStats, error)

CompileFile is Compile with cancellation and statistics.

func (*Builder) CompileTo 

func (b *Builder) CompileTo(w io.Writer) (BuildStats, error)

CompileTo writes the v1 blob for all accumulated rows to w. Nothing is written if duplicate detection or placement fails.

func (*Builder) CompileToContext 

func (b *Builder) CompileToContext(ctx context.Context, w io.Writer) (BuildStats, error)

CompileToContext is CompileTo with cancellation (checked every 64k rows / buckets and during placement). On cancellation nothing has been written.

func (*Builder) CompileV1 

func (b *Builder) CompileV1(outPath string) error

Compile writes the blob to outPath atomically: it writes a temporary file in the same directory, fsyncs it and renames it over outPath. On error outPath is untouched and the temporary file is removed. CompileV1 writes the legacy hash-and-displace format (version 1). Kept for compatibility and tests; new datasets should use Compile.

func (*Builder) CompileWithOptions 

func (b *Builder) CompileWithOptions(outPath string, opts CompileOptions) error

CompileWithOptions writes a version 2 (.attos) file: a PTHash minimal perfect hash, an offset table and the key-prefixed records stored in slot order, optionally in LZ4 blocks. See FORMAT_V2.md.

func (*Builder) Grow 

func (b *Builder) Grow(n int)

Grow preallocates the row index for n more rows. It avoids the transient garbage of repeated slice growth when the row count is known in advance (about 26 bytes per row).

func (*Builder) Len 

func (b *Builder) Len() int

Len returns the number of rows accumulated so far.

func (*Builder) Reset 

func (b *Builder) Reset()

Reset discards all accumulated rows.

type CompileOptions 

type CompileOptions struct {
	Compression Compression
	// BlockSize is the number of records per compressed block: a power of two from 1 to 256.
	// Smaller blocks decompress faster but compress less. Default 8.
	BlockSize int
	// OffsetWidth is the byte width of the offsets section: 0 chooses automatically (4 when the blob is at
	// most 2^32-1 bytes, else 5), 4 forces 32-bit offsets, 5 forces 40-bit offsets (blobs up to 1 TiB).
	OffsetWidth int
	// FingerprintBytes selects lean keys: 0 stores full keys (default); 4 or 8 stores only a fingerprint
	// of that many bytes per record instead of the key (records are fingerprint|value). Lookups verify the
	// fingerprint, so a key that was never added is accepted with probability 2^-32 (4) or 2^-64 (8).
	// Scan then yields the fingerprint in place of the key. See FORMAT_V2.md.
	FingerprintBytes int
}

CompileOptions tunes Compile. The zero value means no compression.

type Compression 

type Compression int

Compression selects how records are stored in a version 2 file. 

const (
	// CompressNone stores records uncompressed: fastest lookups, zero allocations per Get.
	CompressNone Compression = iota
	// CompressLZ4 stores records in small LZ4-compressed blocks. Smaller files, but every lookup
	// decompresses one block (roughly 0.4 microseconds per record in the block) and allocates. See FORMAT_V2.md.
	CompressLZ4
)

type Config 

type Config struct {
	TelemetryEnabled bool `json:"telemetry_enabled"`
}

Config is the server-provided SDK configuration (X-Attos-Config header).

type DuplicateKeyError 

type DuplicateKeyError struct {
	FirstLine, Line int
	Key             string // the duplicated key, truncated to 64 bytes (may be empty)
}

DuplicateKeyError reports a key that appears more than once.

func (*DuplicateKeyError) Error 

func (e *DuplicateKeyError) Error() string

func (*DuplicateKeyError) Unwrap 

func (e *DuplicateKeyError) Unwrap() error

type FeatureError 

type FeatureError struct{ Feature string }

FeatureError is returned when a file passes its CRC but uses a format feature this SDK does not implement (an unknown flag bit, offset width or fingerprint width). It matches ErrUnsupportedFeature with errors.Is. Fail-closed: the file is never partially read.

func (*FeatureError) Error 

func (e *FeatureError) Error() string

func (*FeatureError) Unwrap 

func (e *FeatureError) Unwrap() error

type Field 

type Field struct {
	Name string
	Type FieldType
}

Field names one typed field of a value.

type FieldType 

type FieldType int

FieldType is the type of one field of a Schema. 

const (
	U8 FieldType = iota
	U16
	U32
	U64
	I8
	I16
	I32
	I64
	F32
	F64
	Bool
	String // u16 big-endian length, then UTF-8 bytes
	Bytes  // u16 big-endian length, then bytes
)

type Format 

type Format int

Format selects the input format for the Builder. 

const (
	// FormatAuto detects the format from the first non-whitespace byte:
	// '{' selects NDJSON, anything else CSV (CONTRACTS.md §1.5).
	FormatAuto Format = iota
	// FormatCSV forces RFC 4180 CSV with a header row.
	FormatCSV
	// FormatNDJSON forces newline-delimited JSON objects.
	FormatNDJSON
)

type LatencyBucket 

type LatencyBucket struct {
	UpperNs uint64
	Count   uint64
}

LatencyBucket is one cumulative histogram bucket: Count samples with a latency of at most UpperNs nanoseconds that were not counted in a lower bucket.

type LatencySnapshot 

type LatencySnapshot struct {
	Samples            uint64
	P50, P95, P99, Max uint64 // nanoseconds
}

LatencySnapshot summarises sampled lookup latencies.

type LimitError 

type LimitError struct {
	Line     int
	Field    string // "key" or "value"
	Len, Max int
}

LimitError reports a key or value longer than the v1 format allows.

func (*LimitError) Error 

func (e *LimitError) Error() string

func (*LimitError) Unwrap 

func (e *LimitError) Unwrap() error

type MalformedRowError 

type MalformedRowError struct {
	Line   int
	Reason string
}

MalformedRowError reports a row that cannot be parsed. Line is the 1-based physical line number (the CSV header is line 1).

func (*MalformedRowError) Error 

func (e *MalformedRowError) Error() string

func (*MalformedRowError) Unwrap 

func (e *MalformedRowError) Unwrap() error

type Map 

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

Map is a read-only, memory-mapped Attos blob that can be hot-swapped. All methods are safe for concurrent use. Lookups are lock-free and allocation-free.

func Open 

func Open(path string) (*Map, error)

Open maps path using default OpenConfig (no checksum, no mlock).

func OpenWithConfig 

func OpenWithConfig(path string, cfg OpenConfig) (*Map, error)

OpenWithConfig maps path read-only, validates it and returns a Map. Every failure mode returns a typed error: ErrInvalidMagic, ErrInvalidVersion (*VersionError), ErrCorruptedBinary or ErrChecksumMismatch.

func (*Map) Close 

func (m *Map) Close() error

Close stops all further lookups (they return ErrMapClosed), waits for in-flight lookups to finish and unmaps every mapping, including retired ones still in their grace period. Zero-copy values become invalid when Close returns. Close is idempotent.

func (*Map) DrainLatency 

func (m *Map) DrainLatency() LatencySnapshot

DrainLatency returns sampled lookup-latency percentiles since the previous call and resets the histogram (used by the telemetry reporter).

func (*Map) FormatVersion 

func (m *Map) FormatVersion() uint32

FormatVersion returns the file format version of the live mapping (1 or 2), or 0 if no file is loaded.

func (*Map) Get 

func (m *Map) Get(key []byte) ([]byte, error)

Get performs an O(1) lookup and returns a zero-copy slice of the value. See the package comment above for the validity window; use GetCopy or View to keep the value longer. Zero allocations.

func (*Map) GetCopy 

func (m *Map) GetCopy(key, dst []byte) ([]byte, error)

GetCopy looks up key and appends its value to dst, returning the extended slice. The result never aliases the mapping and stays valid across swaps and Close. Zero allocations when cap(dst) is large enough.

func (*Map) GetMany 

func (m *Map) GetMany(keys [][]byte, out [][]byte) (int, error)

GetMany looks up keys in one call and stores each zero-copy value in out[i] (nil when the key is absent). It returns the number of keys found. len(out) must be at least len(keys). Value slices have the same validity window as Get.

It is cheaper per key than repeated Get on large files: the mapping is pinned once, and for each group of keys the routing-table reads are issued together, then the offset-table reads, so their cache misses overlap instead of stalling one after another. Every key counts as a lookup (absent keys as misses); batch calls are not latency-sampled. Zero allocations.

func (*Map) GetString 

func (m *Map) GetString(key string) (string, error)

GetString performs an O(1) lookup returning a zero-copy string with the same validity window as Get. Zero allocations.

func (*Map) LatencyBuckets 

func (m *Map) LatencyBuckets() []LatencyBucket

LatencyBuckets returns the sampled lookup-latency histogram (1 in 256 lookups) as non-empty buckets in ascending order. Unlike DrainLatency it is never reset, so exporters can read it at any time.

func (*Map) Len 

func (m *Map) Len() int

Len returns the number of keys in the current mapping (0 if closed).

func (*Map) Lookups 

func (m *Map) Lookups() uint64

Lookups returns the total number of lookups served by this Map so far.

func (*Map) Misses 

func (m *Map) Misses() uint64

Misses returns the number of lookups that found no value. Hits are Lookups() - Misses().

func (*Map) PendingRetired 

func (m *Map) PendingRetired() int

PendingRetired returns the number of replaced mappings not yet unmapped.

func (*Map) Reload 

func (m *Map) Reload(path string) error

Reload opens and validates path (honouring the Map's OpenConfig) and atomically hot-swaps it in. In-flight lookups finish on the old mapping, which is unmapped after they drain plus RetireGrace. Reload blocks while two retired mappings are still pending. On error the current map is kept.

func (*Map) Scan 

func (m *Map) Scan(fn func(key, value []byte) bool)

Scan iterates over all entries in slot order (arbitrary, MPHF-defined). The mapping is pinned for the whole scan; key and value slices are valid only during the callback. Return false to stop early.

func (*Map) Size 

func (m *Map) Size() int64

Size returns the size in bytes of the currently mapped blob (0 if closed).

func (*Map) Slots 

func (m *Map) Slots() uint64

Slots returns the number of slots in the live mapping (0 if none).

func (*Map) Swaps 

func (m *Map) Swaps() uint64

Swaps returns how many times the mapping has been replaced.

func (*Map) View 

func (m *Map) View(key []byte, fn func(value []byte)) error

View looks up key and calls fn with a zero-copy value. The mapping is pinned for the duration of fn, so the slice is valid until fn returns (fn must not retain it, call Close or block on a swap).

func (*Map) WritePrometheus 

func (m *Map) WritePrometheus(w io.Writer)

WritePrometheus writes the Prometheus text exposition (version 0.0.4) for the map: lookups, misses and hit ratio, hot swaps, file size and version, and a sampled lookup-latency histogram (1 in 256 lookups). Everything is read from counters the SDK keeps anyway, so scraping adds nothing to the lookup path.

type OpenConfig 

type OpenConfig struct {
	// PinToMemory calls mlock(2) after mapping to prevent the OS from
	// paging out the region. Recommended for latency-critical deployments
	// (GKE Standard supports mlock; Autopilot does not).
	// Default: false. Failure is non-fatal; a warning is emitted via Logger.
	PinToMemory bool

	// ValidateChecksum verifies the CRC32-IEEE trailer before the map is
	// exposed. The Synchronizer always validates downloaded blobs regardless
	// of this flag. Default: false (direct Open only).
	ValidateChecksum bool

	// WarmUp reads every page right after mapping so the first lookups never fault. It makes the whole
	// file resident, so resident memory equals the file size. Default: false: only the pages lookups
	// actually touch stay in memory (with MADV_RANDOM, no readahead).
	WarmUp bool

	// Warm selects how much of the file is read in at open: WarmNone (default) maps lazily, so only the
	// pages lookups touch become resident; WarmIndex reads the header, hash tables and offset table
	// (about 2-4% of the file) so a lookup against a cold page cache faults in one record page instead
	// of three; WarmFull reads every page (same as WarmUp: true).
	Warm WarmMode

	// SkipWarmUp is deprecated and has no effect unless WarmUp is set, in which case it wins.
	SkipWarmUp bool

	// RetireGrace is the minimum time a replaced map stays mapped after a hot
	// swap (in addition to waiting for in-flight lookups to finish). It bounds
	// the validity window of zero-copy values returned by Get/GetString: such
	// a value stays readable for at least RetireGrace after the map it came
	// from was replaced. Default (0): DefaultRetireGrace. Negative: no grace.
	RetireGrace time.Duration

	// Logger is used for non-fatal operational messages (mlock failures,
	// warm-up, etc.). If nil, log.Default() is used.
	Logger *log.Logger

	// LocalObjectStore instructs the Synchronizer to pull directly from a
	// customer S3 bucket instead of polling the Attos control plane.
	LocalObjectStore bool

	// LocalS3Bucket is the name of the customer's S3 bucket to pull from
	// when LocalObjectStore is true. The object key is "<stashID>.attos".
	LocalS3Bucket string
}

OpenConfig controls how a Map is opened and validated. All fields are optional; zero values select safe defaults.

type Option 

type Option func(*Synchronizer)

Option configures the Attos Synchronizer.

func WithAPIKey 

func WithAPIKey(key string) Option

WithAPIKey sets the API key sent in X-API-Key on sync and telemetry requests.

func WithBaseURL 

func WithBaseURL(url string) Option

WithBaseURL sets the control plane base URL. Default: "http://localhost:8080".

func WithCacheDir 

func WithCacheDir(dir string) Option

WithCacheDir sets the directory holding the live blob (<stashID>.attos), its ETag and in-progress downloads (always in the same directory, so the final rename is atomic). Default: "./.attos/cache".

func WithCompression 

func WithCompression(on bool) Option

WithCompression controls whether syncs ask for the gzip copy of the blob (Accept-Encoding: gzip; default true). Compression is transport-only: the stored and memory-mapped blob is always the v1 file.

func WithHTTPClient 

func WithHTTPClient(c *http.Client) Option

WithHTTPClient sets the HTTP client used for sync and telemetry.

func WithLogger 

func WithLogger(l *log.Logger) Option

WithLogger injects a custom *log.Logger for all operational messages emitted by the Synchronizer (sync errors, hot-swap events, telemetry failures). If not set, log.Default() is used.

func WithNodeID 

func WithNodeID(id string) Option

WithNodeID overrides the X-Attos-Node-ID (1–128 chars). Default: the ATTOS_NODE_ID environment variable, else the hostname, else a random UUIDv4.

Active nodes are counted (and billed) by distinct node ID, so every process on one host counts as one node by default; containers sharing a hostname are one node too. Set a distinct ID per process if they should count separately.

func WithOpenConfig 

func WithOpenConfig(cfg OpenConfig) Option

WithOpenConfig sets the OpenConfig for the Synchronizer.

func WithSignatureKeys 

func WithSignatureKeys(keys map[string]ed25519.PublicKey) Option

WithSignatureKeys enforces signed artifacts: only files whose signature verifies against one of these public keys (by key id) are installed.

func WithSyncInterval 

func WithSyncInterval(d time.Duration) Option

WithSyncInterval sets the background polling interval. A value of 0 disables background polling (manual Sync calls only). Default: DefaultSyncInterval (30 s), identical across the Go, Rust and Java SDKs.

func WithTelemetryAPIKey 

func WithTelemetryAPIKey(key string) Option

WithTelemetryAPIKey uses a different API key for telemetry than for sync.

func WithTelemetryBaseURL 

func WithTelemetryBaseURL(url string) Option

WithTelemetryBaseURL sends telemetry to a different base URL than sync.

func WithTelemetryDisabled 

func WithTelemetryDisabled() Option

WithTelemetryDisabled turns the telemetry emitter off regardless of the server's X-Attos-Config.

func WithTelemetryInterval 

func WithTelemetryInterval(d time.Duration) Option

WithTelemetryInterval sets the telemetry report interval (default 60 s). 0 disables the telemetry emitter.

type Schema 

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

Schema describes a structured value (a client-side convention; the file format is unchanged). Fields are laid out in declaration order; integers and floats are big-endian, Bool is one byte (0 or 1), String and Bytes are a u16 big-endian length followed by the content. The Rust and Java SDKs use the same layout.

func NewSchema 

func NewSchema(fields ...Field) *Schema

NewSchema returns a Schema for fields in declaration order.

func (*Schema) Bool 

func (sc *Schema) Bool(v []byte, name string) (bool, error)

Bool reads a Bool field.

func (*Schema) Bytes 

func (sc *Schema) Bytes(v []byte, name string) ([]byte, error)

Bytes reads a Bytes field as a zero-copy slice of v.

func (*Schema) Encode 

func (sc *Schema) Encode(vals ...any) ([]byte, error)

Encode builds a value from vals in declaration order: uint64/int64/float64/ bool/string/[]byte (any integer or float kind is accepted for numeric fields).

func (*Schema) Float 

func (sc *Schema) Float(v []byte, name string) (float64, error)

Float reads an F32 or F64 field.

func (*Schema) Int 

func (sc *Schema) Int(v []byte, name string) (int64, error)

Int reads an I8, I16, I32 or I64 field.

func (*Schema) String 

func (sc *Schema) String(v []byte, name string) (string, error)

String reads a String field; the result aliases v (same validity window as v).

func (*Schema) Uint 

func (sc *Schema) Uint(v []byte, name string) (uint64, error)

Uint reads a U8, U16, U32 or U64 field.

type StatusError 

type StatusError struct {
	Code       int
	RetryAfter time.Duration // parsed Retry-After for 402/429 (0 if absent)
	Body       string
}

StatusError is returned by Sync for a non-200/304 HTTP response.

func (*StatusError) Error 

func (e *StatusError) Error() string

func (*StatusError) Quota 

func (e *StatusError) Quota() bool

Quota reports whether the error is a node/quota limit (402 or 429).

type Synchronizer 

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

Synchronizer is the data-plane SDK for syncing and querying binary blobs locally. It manages the full lifecycle: download, validation, mmap, atomic hot-swap, background polling with back-off, telemetry and shutdown.

Lookups (Get, GetString, GetCopy, View) are lock-free and allocation-free; see Map for the retirement scheme and the zero-copy validity window.

func NewSynchronizer 

func NewSynchronizer(ctx context.Context, stashID string, opts ...Option) (*Synchronizer, error)

NewSynchronizer creates and initialises a Synchronizer for stashID.

It first loads a previously synced blob from the cache directory (if one exists and validates), then performs an initial Sync. If that sync fails but a cached blob is loaded, the error is logged and the cached data is served; with no data at all the error is returned. Background polling runs every WithSyncInterval (default DefaultSyncInterval; 0 disables polling).

func (*Synchronizer) Close 

func (s *Synchronizer) Close() error

Close stops background polling and telemetry, waits for them to exit, then closes the map (waiting for in-flight lookups and unmapping every region). It is safe to call more than once.

func (*Synchronizer) DatasetName deprecated 

func (s *Synchronizer) DatasetName() string

DatasetName returns the stash's display name.

Deprecated: datasets are now called stashes; use StashName.

func (*Synchronizer) ETag 

func (s *Synchronizer) ETag() string

ETag returns the ETag of the currently loaded blob ("" if unknown).

func (*Synchronizer) FlushTelemetry 

func (s *Synchronizer) FlushTelemetry() error

FlushTelemetry builds and enqueues a report immediately (testing/diagnostics).

func (*Synchronizer) Get 

func (s *Synchronizer) Get(key []byte) ([]byte, error)

Get performs an O(1) lock-free, allocation-free lookup. The returned slice aliases the mapping (validity window: see Map). Use GetCopy to keep it.

func (*Synchronizer) GetCopy 

func (s *Synchronizer) GetCopy(key, dst []byte) ([]byte, error)

GetCopy appends the value for key to dst (safe to keep across swaps).

func (*Synchronizer) GetString 

func (s *Synchronizer) GetString(key string) (string, error)

GetString is the zero-copy string variant of Get.

func (*Synchronizer) Map 

func (s *Synchronizer) Map() *Map

Map returns the hot-swappable map handle served by this Synchronizer.

func (*Synchronizer) NodeID 

func (s *Synchronizer) NodeID() string

NodeID returns the X-Attos-Node-ID sent by this process.

func (*Synchronizer) StashName 

func (s *Synchronizer) StashName() string

StashName returns the stash's display name as sent by the control plane (the X-Dataset-Name header, "" if none).

func (*Synchronizer) Sync 

func (s *Synchronizer) Sync(ctx context.Context) error

Sync performs one sync attempt: conditional GET (If-None-Match), and on 200 download → fsync → validate → rename → mmap → atomic publish. On any failure the current map keeps serving. It does not sleep or retry; the background loop applies the back-off policy.

func (*Synchronizer) TelemetryEnabled 

func (s *Synchronizer) TelemetryEnabled() bool

TelemetryEnabled reports the server-provided telemetry switch.

func (*Synchronizer) TelemetryStats 

func (s *Synchronizer) TelemetryStats() TelemetryStats

TelemetryStats returns the emitter counters (zero if telemetry is off).

func (*Synchronizer) View 

func (s *Synchronizer) View(key []byte, fn func([]byte)) error

View calls fn with a zero-copy value while the mapping is pinned.

func (*Synchronizer) WritePrometheus 

func (s *Synchronizer) WritePrometheus(w io.Writer)

WritePrometheus writes the map metrics plus sync state: the ETag and stash of the file in use, CRC failures, sync outcomes and telemetry delivery.

type TelemetryStats 

type TelemetryStats struct {
	Sent, Attempts, DroppedPending uint64
	Queued                         int
}

TelemetryStats exposes emitter counters (for tests and diagnostics).

type VersionError 

type VersionError struct{ Version uint32 }

VersionError is returned when a blob carries an unsupported version. It matches ErrInvalidVersion with errors.Is.

func (*VersionError) Error 

func (e *VersionError) Error() string

func (*VersionError) Unwrap 

func (e *VersionError) Unwrap() error

type WarmMode 

type WarmMode int

WarmMode selects how much of a file OpenConfig reads in at open. 

const (
	// WarmNone maps the file lazily. Lowest resident memory; a lookup against a cold page cache pays
	// one disk read per page it touches until the working set is cached.
	WarmNone WarmMode = iota
	// WarmIndex reads in everything before the record data: header, hash tables and offset table.
	WarmIndex
	// WarmFull reads in the whole file. Resident memory equals the file size.
	WarmFull
)

Source Files

View all Source files

Directories

Path Synopsis
cmd
attos-size command
Command attos-size estimates what a dataset costs to serve from Attos compared with Redis: it samples your data (a CSV in Attos input format, a key,value CSV export, or a live Redis read-only), builds real Attos files from the sample, and prints the projected file sizes, RAM saved and the equivalent Redis monthly cost.
 Command attos-size estimates what a dataset costs to serve from Attos compared with Redis: it samples your data (a CSV in Attos input format, a key,value CSV export, or a live Redis read-only), builds real Attos files from the sample, and prints the projected file sizes, RAM saved and the equivalent Redis monthly cost.
internal
binary
hash
lz4
Package lz4 is a small, dependency-free implementation of the LZ4 block format (https://github.com/lz4/lz4/blob/dev/doc/lz4_Block_format.md): a greedy compressor used when the SDK builds compressed .attos files, and a bounds-checked decompressor used by readers.
 Package lz4 is a small, dependency-free implementation of the LZ4 block format (https://github.com/lz4/lz4/blob/dev/doc/lz4_Block_format.md): a greedy compressor used when the SDK builds compressed .attos files, and a bounds-checked decompressor used by readers.
mmap
pthash
Package pthash implements the minimal perfect hash used by the .attos v2 format (PTHash, Pibiri & Trani 2021, "dual compact" pilot encoding).
 Package pthash implements the minimal perfect hash used by the .attos v2 format (PTHash, Pibiri & Trani 2021, "dual compact" pilot encoding).

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.