blob

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Jul 13, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

Documentation

Overview

Package blob provides claim-check values and content-addressed storage for large Temporal workflow payloads.

Temporal workflows can quickly hit history size limits when handling large data, such as ML models, agentic conversation context, or large documents. This package solves this by keeping large data out of the workflow history and letting you pass typed handles between activities instead.

Storage gives one backend two related uses. First, it serves as Temporal's native external storage driver so large automatic SDK payloads are kept out of history. Second, it provides the explicit typed Value for application data that should be written once and passed around by handle.

A Value of type T carries exactly one of an inline typed value or a Ref to bytes stored in a Backend. Activity code can create values with Offload and read them with Value.Resolve or Resolve. Workflow code stays I/O-free: it creates inline values with Inline, passes Value handles between activities, and can optionally materialize an existing Ref through GetValue. GetValue schedules the package's blob.get activity, so the worker must register blob activities and the Temporal client must configure external storage for large activity results.

Wiring storage

Storage is the startup handle for an application process. If you run your own Temporal client and worker, create one Storage value from your Backend, pass Storage.ExternalStorage to client.Options.ExternalStorage, pass Storage.WorkerInterceptor to worker.Options.Interceptors, and call Storage.RegisterActivities on each worker that may execute blob.get. Then register your normal workflows and activities on that same worker. If you use embeddedtemporal, embeddedtemporal.WithLocalStorage performs the client, worker, and local backend wiring for servers started through that package.

Explicit blob value bytes are Temporal payload envelopes produced by the Store's configured DataConverter. Pass WithDataConverter the same DataConverter configured on the Temporal client so blob bytes and history payloads use the same encoding and codec layers.

Backend and owner metadata

Backend implementations store immutable bytes by their SHA-256 content ID and maintain owner edges from blobs to workflow executions. Normal workflow and activity code should not need to populate Owner directly: activity-side Offload and Temporal ExternalStorage derive it from the Temporal execution. Backend authors must persist it because future garbage collection can only safely delete blobs after the owning workflow edges are gone. ContextWithOwner exists for tests and advanced explicit writes outside a Temporal activity.

For a full multi-turn workflow walkthrough, see https://vihren.dev/vihren/examples/conversationstorage.

Index

Examples

Constants

View Source 
const (
	// BlobGetActivityName is the derived Temporal activity type.
	BlobGetActivityName = "blob.get"
)
View Source 
const (
	// DefaultInlineThreshold is the platform default maximum serialized byte
	// length for values that are allowed to cross workflow history inline.
	DefaultInlineThreshold = 256 << 10
)

Variables

View Source 
var (
	// ErrInvalidID reports that a string is not a canonical blob ID.
	ErrInvalidID = errors.New("blob: invalid blob ID")

	// ErrInvalidValue reports that a Value has neither or both union arms set.
	ErrInvalidValue = errors.New("blob: invalid value")

	// ErrInlineTooLarge reports that an inline-only encode exceeded its threshold.
	ErrInlineTooLarge = errors.New("blob: inline value exceeds threshold")

	// ErrNotFound reports that no blob exists for the requested ID.
	ErrNotFound = errors.New("blob: blob not found")

	// ErrNoStore reports that an operation needs an ambient store in the context.
	ErrNoStore = errors.New("blob: store is not available in context")

	// ErrStoreRequired reports that an explicit blob operation needs a Store.
	ErrStoreRequired = errors.New("blob: store is required")

	// ErrBackendRequired reports that a Store has no usable backend.
	ErrBackendRequired = errors.New("blob: backend is required")

	// ErrOwnerRequired reports that a write did not identify an owning workflow.
	ErrOwnerRequired = errors.New("blob: owner workflow is required")

	// ErrDigestMismatch reports that stored bytes do not hash to their ID.
	ErrDigestMismatch = errors.New("blob: digest mismatch")

	// ErrInvalidConfig reports that backend-neutral store configuration is not
	// usable.
	ErrInvalidConfig = errors.New("blob: invalid config")
)
View Source 
var Activity activityProxy

Activity is this package's generated activity proxy.

Functions

func ContextWithOwner 

func ContextWithOwner(ctx context.Context, owner Owner) context.Context

ContextWithOwner installs an explicit owner for tests and advanced explicit writes that are not running inside a Temporal activity. Production activity code normally relies on OwnerFromActivity instead.

func ContextWithStore 

func ContextWithStore(ctx context.Context, store *Store) context.Context

ContextWithStore installs a Store into ctx for activity-side Resolve and Offload calls. Workers normally install this with the storage interceptor.

func Get 

func Get[T any](ctx context.Context, store *Store, ref Ref[T]) (T, error)

Get fetches, digest-verifies, and deserializes the blob referenced by ref with store's DataConverter.

func GetValue 

func GetValue[T any](ctx workflow.Context, ref Ref[T]) (T, error)

GetValue resolves ref from workflow code by scheduling BlobGetActivity and decoding the returned value through the calling workflow context's configured DataConverter.

The worker must register blob activities with Storage.RegisterActivities, and the Temporal client must configure ExternalStorage for large activity results. Configure blob.get routing and timeouts with workflow.WithActivityOptions on ctx. If ctx has neither StartToCloseTimeout nor ScheduleToCloseTimeout, GetValue adds a default StartToCloseTimeout. The configured DataConverter must support converter.RawValue; Temporal's default converter and codec-wrapped default converter do.

func NewWorkerInterceptor 

func NewWorkerInterceptor(store *Store) interceptor.WorkerInterceptor

NewWorkerInterceptor constructs the worker interceptor that injects store into every activity context.

func Register 

func Register(r worker.Registry, activities *Activities)

Register registers this package's generated activities and workflows.

func RegisterActivities 

func RegisterActivities(r activityRegistry, activities *Activities)

RegisterActivities registers this package's generated activities.

func RegisterWorkflows 

func RegisterWorkflows(r worker.Registry)

RegisterWorkflows registers this package's generated workflows.

func Resolve 

func Resolve[T any](ctx context.Context, store *Store, value Value[T]) (T, error)

Resolve returns the inline value or fetches and decodes the referenced value from store.

Types

type Activities 

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

Activities groups blob activity implementations that hold process-local backend dependencies.

func NewActivities 

func NewActivities(store *Store) *Activities

NewActivities constructs blob activities backed by store.

func (*Activities) BlobGetActivity 

func (activities *Activities) BlobGetActivity(ctx context.Context, id ID) (converter.RawValue, error)

BlobGetActivity fetches, digest-verifies, and returns the raw stored payload for id with the store converter's codec layer stripped. Temporal's result encoder then applies the caller's configured codec layer exactly once.

type Backend 

type Backend interface {
	Put(ctx context.Context, data []byte, meta WriteMeta) (ID, error)
	Get(ctx context.Context, id ID) ([]byte, error)
	Exists(ctx context.Context, id ID) (bool, error)
	AddOwner(ctx context.Context, id ID, owner Owner) error
	RemoveOwner(ctx context.Context, id ID, owner Owner) error
}

Backend stores immutable content-addressed bytes and the workflow owner edges that make those bytes reclaimable by a future GC.

Implementations derive the returned ID from the bytes passed to Put; callers do not supply object names. Put must be idempotent for the same data and must still record the WriteMeta.Owner edge when the bytes already exist. Get returns the exact bytes for an ID or ErrNotFound when the object is absent. Exists reports object presence without reading the bytes.

AddOwner and RemoveOwner maintain the metadata graph from blob IDs to owning workflows. They are separate from Put so a backend can attach additional workflow owners to already stored content. AddOwner should return ErrNotFound when the blob is missing. RemoveOwner deletes only that owner edge; it must not delete the blob bytes by itself. Automatic garbage collection is not implemented yet, but backend metadata should be shaped so GC can later list blobs whose owner set is empty.

type ID 

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

ID identifies immutable blob bytes by SHA-256 content digest. Its canonical string and JSON form is "sha256:<64 lowercase hex characters>".

The zero value is a well-formed all-zero digest. It is not assumed to exist in storage; normal reads return ErrNotFound when no bytes were written for it.

func ParseID 

func ParseID(s string) (ID, error)

ParseID validates and returns a canonical sha256 content ID.

func (ID) Digest 

func (id ID) Digest() [sha256.Size]byte

Digest returns the SHA-256 digest bytes represented by id.

func (ID) MarshalJSON 

func (id ID) MarshalJSON() ([]byte, error)

MarshalJSON encodes id as its canonical string wire form.

func (ID) String 

func (id ID) String() string

String returns the canonical "sha256:<hex>" representation of id.

func (*ID) UnmarshalJSON 

func (id *ID) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes and validates id from its canonical string wire form.

type LocalBackend 

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

LocalBackend stores blobs on the local filesystem and owner edges in SQLite. It is intended for local development, examples, and deterministic tests.

func NewLocalBackend 

func NewLocalBackend(root string) (*LocalBackend, error)

NewLocalBackend constructs a filesystem backend rooted at root. Blob bytes live under root/algorithm/digest and owner metadata lives in root/metadata.db.

func (*LocalBackend) AddOwner 

func (backend *LocalBackend) AddOwner(ctx context.Context, id ID, owner Owner) error

AddOwner records an additional owning workflow edge for an existing blob.

func (*LocalBackend) Close 

func (backend *LocalBackend) Close() error

Close releases the backend metadata database.

func (*LocalBackend) Exists 

func (backend *LocalBackend) Exists(ctx context.Context, id ID) (bool, error)

Exists reports whether id is present in the local store.

func (*LocalBackend) Get 

func (backend *LocalBackend) Get(ctx context.Context, id ID) ([]byte, error)

Get reads the blob stored under id.

func (*LocalBackend) Put 

func (backend *LocalBackend) Put(ctx context.Context, data []byte, meta WriteMeta) (ID, error)

Put stores data by derived content ID and records the owning workflow edge.

func (*LocalBackend) RemoveOwner 

func (backend *LocalBackend) RemoveOwner(ctx context.Context, id ID, owner Owner) error

RemoveOwner removes one owning workflow edge. Bytes are retained; automatic garbage collection is not implemented yet.

type Owner 

type Owner struct {
	Namespace    string `json:"namespace"`
	WorkflowID   string `json:"workflow_id"`
	RunID        string `json:"run_id,omitempty"`
	WorkflowType string `json:"workflow_type,omitempty"`
}

Owner identifies the workflow execution responsible for a blob write.

Normal activity and workflow code usually does not construct Owner values. Activity-side writes derive the owner from the current Temporal activity, and Temporal native ExternalStorage derives it from the SDK storage target. The liveness key is Namespace plus WorkflowID; RunID and WorkflowType are recorded for debugging and operator visibility.

Backend implementations persist owner edges so future garbage collection can distinguish blobs still reachable from live workflows from blobs that may be deleted. Use ContextWithOwner for tests or explicit writes outside Temporal.

func OwnerFromActivity 

func OwnerFromActivity(ctx context.Context) Owner

OwnerFromActivity builds the blob owner from the current Temporal activity.

func OwnerFromTarget 

func OwnerFromTarget(target converter.StorageDriverTargetInfo) (Owner, error)

OwnerFromTarget builds the blob owner from a Temporal external-storage target. Non-workflow targets cannot own blobs and return ErrOwnerRequired.

type Ref 

type Ref[T any] struct {
	ID   ID    `json:"id"`
	Size int64 `json:"size"`
}

Ref is a small, history-safe handle to a stored blob. Size is advisory metadata for fetch/inline decisions; identity and integrity remain the digest.

func Put 

func Put[T any](ctx context.Context, store *Store, value T) (Ref[T], error)

Put serializes value with store's DataConverter, hashes the encoded bytes, stores them, and returns a typed reference. The owner is recovered from a Temporal activity context or ContextWithOwner.

func (*Ref[T]) UnmarshalJSON 

func (ref *Ref[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes a Ref and verifies the required ref shape at the wire boundary before a zero-value ID can be mistaken for an omitted ID.

type Reference 

type Reference struct {
	ID     ID
	Size   int64
	Owners []Owner
}

Reference describes one stored blob and its accumulated owners. Backends may use this shape for admin or GC scans; application code normally passes typed Ref[T] values instead.

type Storage 

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

Storage is the startup handle for one blob backend.

Use one Storage value to wire the same backend into Temporal native ExternalStorage, activity-side Value[T] offload/resolution, and workflow-side GetValue reads. Non-embedded Temporal applications normally pass Storage.ExternalStorage to client.Options.ExternalStorage, append Storage.WorkerInterceptor to worker.Options.Interceptors, and call Storage.RegisterActivities on every worker that may execute blob.get.

Example (NonEmbeddedTemporal)

ExampleStorage_nonEmbeddedTemporal shows the wiring needed when the application owns its Temporal client and worker. 

package main

import (
	"context"
	"log"
	"time"

	"go.temporal.io/sdk/client"
	"go.temporal.io/sdk/converter"
	"go.temporal.io/sdk/interceptor"
	"go.temporal.io/sdk/worker"
	"go.temporal.io/sdk/workflow"
	"vihren.dev/vihren/blob"
)

func main() {
	const taskQueue = "agent-task-queue"

	// Use the same converter configuration for Temporal history and explicit
	// blob.Value bytes. If your app wraps the default converter with encryption,
	// compression, or another codec, use that same configured converter here.
	dataConverter := converter.GetDefaultDataConverter()

	// Storage owns one backend and exposes the three integration points below:
	// ExternalStorage for SDK payload offload, WorkerInterceptor for ambient
	// activity helpers, and RegisterActivities for workflow-side blob.GetValue.
	storage, err := blob.NewLocalStorage(
		"/var/lib/vihren/blob",
		blob.WithDataConverter(dataConverter),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := storage.Close(); err != nil {
			log.Fatal(err)
		}
	}()

	temporalClient, err := client.Dial(client.Options{
		HostPort:      "localhost:7233",
		Namespace:     "default",
		DataConverter: dataConverter,
		// ExternalStorage lets the Temporal SDK replace oversized payloads in
		// history with storage references. This is independent of blob.Value; it
		// protects normal workflow/activity arguments and results too.
		ExternalStorage: storage.ExternalStorage(),
	})
	if err != nil {
		log.Fatal(err)
	}
	defer temporalClient.Close()

	temporalWorker := worker.New(temporalClient, taskQueue, worker.Options{
		// The interceptor puts storage.Store() into every activity context. That
		// is what lets application activities call blob.Offload(ctx, value) and
		// value.Resolve(ctx) without taking *blob.Store as an explicit parameter.
		Interceptors: []interceptor.WorkerInterceptor{storage.WorkerInterceptor()},
	})

	// RegisterActivities installs Vihren's blob.get activity. Workflow code uses
	// that activity when it calls blob.GetValue(ctx, ref). The activity is
	// registered with storage.Store(), so GetValue does not depend on the
	// WorkerInterceptor above.
	storage.RegisterActivities(temporalWorker)

	// Register the application's own workflow and activities after the storage
	// hooks are in place. These activities can use ambient blob helpers because
	// the worker was created with storage.WorkerInterceptor().
	temporalWorker.RegisterWorkflow(storageExampleWorkflow)
	temporalWorker.RegisterActivity(storageExampleProduce)
	temporalWorker.RegisterActivity(storageExampleConsume)

	if err := temporalWorker.Run(worker.InterruptCh()); err != nil {
		log.Fatal(err)
	}
}

// storageExampleWorkflow demonstrates that workflow code passes Value handles
// between activities and does not perform blob I/O itself.
func storageExampleWorkflow(ctx workflow.Context, message string) (string, error) {
	activityCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: time.Minute,
	})

	var stored blob.Value[string]
	if err := workflow.ExecuteActivity(activityCtx, storageExampleProduce, message).Get(activityCtx, &stored); err != nil {
		return "", err
	}

	var resolved string
	if err := workflow.ExecuteActivity(activityCtx, storageExampleConsume, stored).Get(activityCtx, &resolved); err != nil {
		return "", err
	}
	return resolved, nil
}

// storageExampleProduce writes a claim-check value from activity code.
func storageExampleProduce(ctx context.Context, message string) (blob.Value[string], error) {
	return blob.Offload(ctx, message)
}

// storageExampleConsume resolves a claim-check value from activity code.
func storageExampleConsume(ctx context.Context, message blob.Value[string]) (string, error) {
	return message.Resolve(ctx)
}

func NewLocalStorage 

func NewLocalStorage(root string, opts ...StorageOption) (*Storage, error)

NewLocalStorage constructs a Storage handle backed by a local filesystem root.

func NewStorage 

func NewStorage(backend Backend, opts ...StorageOption) (*Storage, error)

NewStorage constructs a Storage handle over backend.

func (*Storage) Backend 

func (storage *Storage) Backend() Backend

Backend returns the underlying backend.

func (*Storage) Close 

func (storage *Storage) Close() error

Close releases closeable storage resources.

func (*Storage) ExternalStorage 

func (storage *Storage) ExternalStorage() converter.ExternalStorage

ExternalStorage returns the Temporal SDK ExternalStorage configuration to set on client.Options.ExternalStorage. This lets the SDK offload oversized Temporal payloads to the same backend used by explicit Value[T] references.

func (*Storage) RegisterActivities 

func (storage *Storage) RegisterActivities(r activityRegistry)

RegisterActivities registers the package's blob.get activity on a worker. It forwards to the generated RegisterActivities; it takes the same minimal activityRegistry surface so any caller-defined registry (not just worker.Registry) can be passed directly instead of re-deriving this wiring by hand.

func (*Storage) Store 

func (storage *Storage) Store() *Store

Store returns the explicit Value[T] store used by Put, Get, and activity side Offload.

func (*Storage) WorkerInterceptor 

func (storage *Storage) WorkerInterceptor() interceptor.WorkerInterceptor

WorkerInterceptor returns the activity interceptor to add to worker.Options.Interceptors. It injects the store used by activity-side Offload and Value.Resolve.

type StorageOption 

type StorageOption func(*storageConfig)

StorageOption customizes explicit blob storage construction.

func WithDataConverter 

func WithDataConverter(dataConverter converter.DataConverter) StorageOption

WithDataConverter overrides the Temporal DataConverter used for explicit blob value bytes.

Pass the same converter configuration to client.Options.DataConverter and WithDataConverter so blob-stored Value[T] bytes and Temporal history payloads use identical encoding and codec layers. The default converter is used when this option is omitted or passed nil.

type Store 

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

Store serializes typed values and stores their bytes by content ID.

func MustNewLocalStore 

func MustNewLocalStore(root string, opts ...StorageOption) *Store

MustNewLocalStore constructs a local store or panics. It is intended for tests and small demos where startup failure is unrecoverable.

func New 

func New(backend Backend, opts ...StorageOption) *Store

New constructs a Store backed by the supplied backend.

func NewLocalStore 

func NewLocalStore(root string, opts ...StorageOption) (*Store, error)

NewLocalStore constructs an explicit Value[T] store backed by a local filesystem root.

func StoreFromContext 

func StoreFromContext(ctx context.Context) (*Store, error)

StoreFromContext returns the ambient Store installed in ctx.

func (*Store) Backend 

func (store *Store) Backend() Backend

Backend returns the store's underlying backend.

type TemporalStorageDriver 

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

TemporalStorageDriver stores Temporal SDK payloads in a blob backend and records workflow owner edges for every offloaded payload.

func NewTemporalStorageDriver 

func NewTemporalStorageDriver(name string, backend Backend) (*TemporalStorageDriver, error)

NewTemporalStorageDriver constructs a native Temporal StorageDriver over the same backend used by explicit Value[T] paths.

func (*TemporalStorageDriver) Name 

func (driver *TemporalStorageDriver) Name() string

Name returns the stable driver instance name persisted in Temporal history.

func (*TemporalStorageDriver) Retrieve 

func (driver *TemporalStorageDriver) Retrieve(
	ctx converter.StorageDriverRetrieveContext,
	claims []converter.StorageDriverClaim,
) ([]*commonpb.Payload, error)

Retrieve fetches and decodes payloads by the content-addressed IDs stored in StorageDriverClaim.ClaimData.

func (*TemporalStorageDriver) Store 

func (driver *TemporalStorageDriver) Store(
	ctx converter.StorageDriverStoreContext,
	payloads []*commonpb.Payload,
) ([]converter.StorageDriverClaim, error)

Store persists payload bytes and records a workflow owner edge.

func (*TemporalStorageDriver) Type 

func (driver *TemporalStorageDriver) Type() string

Type returns the stable implementation type for this driver.

type Value 

type Value[T any] struct {
	Inline *T      `json:"inline,omitempty"`
	Ref    *Ref[T] `json:"ref,omitempty"`
}

Value carries exactly one of an inline typed value or a reference to a stored typed value for activity inputs and outputs that may cross Temporal history.

func Inline 

func Inline[T any](value T) Value[T]

Inline constructs a Value whose data travels inline.

func Offload 

func Offload[T any](ctx context.Context, value T) (Value[T], error)

Offload serializes value with the ambient activity store's DataConverter and returns it inline when the encoded bytes fit DefaultInlineThreshold, or as a stored ref otherwise. It requires a store in ctx even when the value remains inline; workers normally install that store with Storage.WorkerInterceptor.

Example 
package main

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

	"vihren.dev/vihren/blob"
)

func main() {
	root, err := os.MkdirTemp("", "vihren-blob-example-*")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = os.RemoveAll(root) }()

	backend, err := blob.NewLocalBackend(root)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := backend.Close(); err != nil {
			log.Fatal(err)
		}
	}()

	store := blob.New(backend)
	ctx := blob.ContextWithStore(context.Background(), store)
	ctx = blob.ContextWithOwner(ctx, blob.Owner{
		Namespace:  "default",
		WorkflowID: "example-workflow",
	})

	largeValue := strings.Repeat("agent-state-", blob.DefaultInlineThreshold/len("agent-state-")+1)
	stored, err := blob.Offload(ctx, largeValue)
	if err != nil {
		log.Fatal(err)
	}

	resolved, err := stored.Resolve(ctx)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(stored.Ref != nil)
	fmt.Println(len(resolved) == len(largeValue))

}

Output:
true
true

func OffloadWithThreshold 

func OffloadWithThreshold[T any](ctx context.Context, value T, threshold int) (Value[T], error)

OffloadWithThreshold is Offload with a caller-supplied serialized byte threshold. It requires an ambient store in ctx before it can measure the encoded value.

func RefValue 

func RefValue[T any](ref Ref[T]) Value[T]

RefValue constructs a Value whose data travels by blob reference.

func Zero 

func Zero[T any]() Value[T]

Zero returns the zero Value[T].

func (Value[T]) MarshalJSON 

func (value Value[T]) MarshalJSON() ([]byte, error)

MarshalJSON encodes Value's inline-or-ref union. The zero value encodes as null so error paths can carry zero result values without minting malformed inline/ref objects.

func (Value[T]) Resolve 

func (value Value[T]) Resolve(ctx context.Context) (T, error)

Resolve returns the inline value or fetches the referenced value from the ambient activity store in ctx.

func (*Value[T]) UnmarshalJSON 

func (value *Value[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes Value's inline-or-ref union and rejects object values with both arms or neither arm present. JSON null decodes to the zero Value for Temporal result slots on failing activity/workflow executions.

type WriteMeta 

type WriteMeta struct {
	Owner Owner
}

WriteMeta is the mandatory write-time context for backend writes. Backend implementations must reject writes with an empty Owner by returning ErrOwnerRequired.

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.