exporthistory

package
v1.0.0-beta2 Latest Latest
Warning

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

Go to latest
Published: Sep 29, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

README

History export (preview)

exporthistory exports terminal orchestration histories out of a task hub and into Azure Blob Storage. It is the Go counterpart of the .NET Microsoft.DurableTask.ExportHistory preview package, but its export-job lifecycle protocol is Go-specific. Mixed .NET or older Go export-history workers are not supported in the same task hub.

Maturity: preview. The exported API, the serialized shape of the ExportJob entity state, and the names of the system entity, orchestrators, and activities may change in a future release without a major version bump. Drain and delete export jobs before upgrading. Legacy export-job state is not supported or migrated.

Model

An export job is a durable entity keyed by job ID.

Component Name Role
Entity ExportJob Owns lifecycle, configuration, checkpoint, and progress
Orchestrator ExecuteExportJobOperationOrchestrator Runs one entity operation on behalf of a client
Orchestrator ExportJobOrchestrator Lists, exports, and checkpoints; one per run generation
Activity ListTerminalInstancesActivity Returns one page of terminal instance IDs
Activity ExportInstanceHistoryActivity Exports one instance's history

Creating a job moves the entity to Active, mints a run token, and reserves the generation-specific instance ID ExportJob-<jobId>-<runToken> before signaling Run. Run starts ExportJobOrchestrator using that stored ID. Read description.OrchestratorInstanceID to identify the current generation; a job ID alone no longer identifies an orchestration. The orchestration repeatedly lists terminal instances matching the job's filter, exports each instance's history in bounded parallel windows, and commits a checkpoint back to the entity. A batch job completes when the task hub reports no more pages; a continuous job idles for a minute and lists again.

Progress counters

ScannedInstances and ExportedInstances are cumulative processing totals, not counts of distinct instances or executions. This retains the Python SDK's counter semantics. A continuous job preserves the last opaque cursor when a terminal page has no next token. If the source returns that page again, its entries and successful exports are counted again, even when no new instance has completed. Retries of blob writes remain idempotent, but that does not make these counters unique. Do not use them as exact unique-export or billing totals.

Lifecycle
Operation From To
Create Pending, Failed, Completed Active
MarkAsCompleted Active Completed
MarkAsFailed Active Failed

Every other transition raises an InvalidTransitionError, including recreating a job that is already Active. Recreating a Failed or Completed job resets its progress counters and checkpoint but preserves its original creation time. Each successful Create assigns a fresh orchestration ID, so concurrent creates are resolved entirely by the entity's lifecycle transition: only one generation can be Active at a time, and creation performs no destructive cleanup. Previous generations' orchestration histories are retained, not purged on recreation.

A page whose exports keep failing commits without a checkpoint and with the collected failures, which implicitly moves the job to Failed and leaves the cursor on the failing page so a fix can resume from it. That implicit failure goes through the same lifecycle transition as an explicit MarkAsFailed.

The entity's Delete atomically captures its current orchestration ID, clears the state, and returns that ID as a string (an empty string if the job is absent). The client then terminates and purges only that captured generation. State deletion and orchestration cleanup are not atomic, but a Create between them is safe: cleanup never looks up or targets the replacement. A result decoding failure is reported without attempting cleanup.

Deleting an absent job is a no-op. A separate Delete that executes after recreation legitimately removes the new generation. If cleanup fails, the removed entity state stays deleted.

Run fencing

Each Create mints a random run token that is stored on the job and carried by the run it starts. Every orchestration-originated mutation — CommitCheckpoint, MarkAsCompleted, MarkAsFailed, and the Run signal — carries that token, and the entity requires a matching nonempty token. A run left over from a job that was deleted and recreated, or from a prior in-place recreate, therefore cannot checkpoint, complete, or fail the new job. The run also stops itself as soon as it reads a job whose token no longer matches.

Missing tokens never match, even when both the request and stored token are empty. Runs do not adopt another generation's token. A delayed Run signal is dropped after deletion or recreation, and an already-emitted delayed Start stops when it reads the missing or replacement job instead of exporting its data.

There is one current state-owning generation, not a guarantee of zero overlap. Previous orchestrations and already-scheduled activities can briefly remain in flight, but cannot mutate the replacement's job state. Recreation does not purge their histories; Delete cleans only the generation it removes.

Output layout

Each exported instance becomes one object named <prefix><sha256(completedAt|instanceId)>.<extension>:

Format Extension Content type Content-Encoding
ExportFormatJSONL (default) jsonl.gz application/gzip not set
ExportFormatJSON json application/json not set

A JSONL object is gzip-compressed and stored as an opaque gzip file: the name and content type agree and no content coding is declared, so every reader downloads exactly the gzip stream the name promises. Declaring the compression as Content-Encoding: gzip instead would let some clients transparently decompress the download while others would not, leaving a reader unable to tell what it received.

JSONL objects carry one api.HistoryEvent per line; JSON objects contain an array of the same events. Azure objects carry schemaVersion, instanceIdBase64, and executionIdBase64 metadata. The identifier values are always UTF-8 bytes encoded as unpadded RFC 4648 base64url (base64.RawURLEncoding in Go), even for ASCII identifiers. Decode those suffixed keys to recover the original IDs. Raw identifiers are never sent as Azure metadata headers; the history body and object-name hash still use the original IDs. Metadata keys may be returned with different casing by Azure. The name is derived deterministically, so re-exporting an instance overwrites its object instead of duplicating it.

When no destination is supplied, a job writes to the client's configured container under the prefix <mode>-<jobId>/.

Worker setup

store, err := exporthistory.NewAzureBlobHistoryStore(exporthistory.AzureBlobHistoryStoreOptions{
    ConnectionString: storageConnectionString,
    ContainerName:    "history-exports",
})
err = exporthistory.Register(registry, exporthistory.WorkerOptions{
    Source: taskHubClient,
    Store:  store,
})
worker, err := durabletaskscheduler.NewWorker(options, registry, logger,
    durabletaskclient.WithAutoWorkItemFilters(),
    exporthistory.WithExportHistory(),
)

Source supplies the three management reads the export performs: ListInstanceIDs, FetchOrchestrationMetadata, and StreamOrchestrationHistory. *client.TaskHubGrpcClient and the Durable Task Scheduler client satisfy it.

The exporter pins the read to metadata's execution ID and incrementally checks HistoryQuery.MaxEvents, the aggregate approximate byte limit, and the observed ExecutionStarted identity. It does not retain an event list or a whole serialized/compressed history. Per export, memory consists of the current source chunk/event, its encoded JSON, gzip state when needed, a single 1 MiB upload block, and Azure's bounded block-ID list (at most 50,000 entries). Parallel exports each own that budget; large individual events/source chunks still require memory.

Store is a narrow interface with a single Write method. AzureBlobHistoryStore is the production implementation; supply your own to export elsewhere. It is deliberately separate from payload.AzureBlobStore, whose large-payload contract assigns random object names inside a single container. Its endpoint validation is at least as strict as that store's: an AccountURL carrying userinfo, a query string, or a fragment is rejected outright, and plaintext HTTP is confined to loopback endpoints behind AllowInsecureHTTP.

ExportObject.Content is now a single-use io.Reader, not []byte. Custom stores must consume it synchronously, honor cancellation, and publish only after a clean EOF: a read error must leave an existing object unchanged. The Azure implementation stages blocks sequentially and commits the block list only after successful source validation and gzip finalization. On source or upload failure, the opposite side is canceled and joined. SDK transport retries rewind only the current block; an activity retry opens a fresh pinned history stream, never reuses a consumed reader. Failed attempts can leave uncommitted Azure blocks for service-managed expiration, but do not replace committed data. Custom history sources must deliver events serially and stop on handler errors or context cancellation. Update preview source/store implementations and blob metadata readers when upgrading.

Versioning

Every system task is registered unversioned so it stays reachable when an application enables default versioning. Because a strict worker advertises its own version for unversioned registrations, WithExportHistory() allow-lists both the system orchestrators and their activities, so the derived work-item filters keep advertising them unversioned. Without it, a strict-version worker constructs successfully but the service never dispatches export work to it.

Client

exportClient, err := exporthistory.NewClient(taskHubClient, exporthistory.ClientOptions{
    ContainerName: "history-exports",
})

job, err := exportClient.CreateJob(ctx, exporthistory.JobCreationOptions{
    Mode:                 exporthistory.ExportModeBatch,
    CompletedTimeFrom:    from,
    CompletedTimeTo:      to,
    MaxInstancesPerBatch: 200,
})
description, err := job.Describe(ctx)
page, err := exportClient.ListJobs(ctx, exporthistory.ExportJobQuery{JobIDPrefix: "nightly-"})
err = job.Delete(ctx)
Validation

JobCreationOptions.Normalize applies the same rules as .NET:

  • Batch mode requires CompletedTimeFrom and CompletedTimeTo, requires CompletedTimeTo to be strictly greater than CompletedTimeFrom, and rejects an upper bound in the future.
  • Continuous mode rejects CompletedTimeTo and defaults CompletedTimeFrom to now.
  • MaxInstancesPerBatch must be between 1 and 1000; it defaults to 100.
  • RuntimeStatus accepts only COMPLETED, FAILED, and TERMINATED; an empty filter selects all three.
  • A missing job ID is generated as a 32-character GUID.

The ExportJob entity re-validates creation options on the worker, whose clock is independent of the client's. Everything is checked identically except the "upper bound is not in the future" rule, which the entity relaxes by MaxCreationClockSkew (5 minutes) so a worker running slightly behind does not reject a window the client accepted. Clients stay strict, and the tolerance never shifts the window a continuous job starts from.

Errors
Error Sentinel Raised when
ValidationError ErrValidation Invalid options, destination, or client configuration
NotFoundError ErrJobNotFound Reading a job that does not exist
InvalidTransitionError ErrJobInvalidTransition An operation the lifecycle does not allow
OperationError ErrJobOperationFailed An operation orchestration failed for another reason

Errors raised inside the entity carry stable cross-language error types, so a client reconstructs the typed error across the orchestration boundary and errors.Is/errors.As keep working.

Service behavior and limitations

  • Extended sessions are not supported.
  • Pagination. DTS signals the end of the stream by omitting the continuation token. An empty page that still carries a token advances the cursor and the job keeps listing. If a non-empty final page omits its token, a continuous job re-scans that page on each idle cycle; deterministic blob names make the writes idempotent, but scanned and exported counters include the re-scan.
  • List visibility lag. The service's instance-ID index can lag orchestration completion. A batch job whose first page is empty legitimately completes with nothing exported, so schedule a job after the window's instances are listable.
  • Retry semantics. A per-instance export is attempted up to three times (retry delays 15s, then 30s) for transient failures such as a storage write error. Conditions retrying cannot fix, such as a missing or non-terminal instance, are collected immediately. A page whose instances still fail is attempted three times in total, waiting 1 minute before the second attempt and 2 minutes before the third; the third attempt fails the page instead of waiting again.
  • Delete cleanup is separate from entity deletion and targets only the captured generation. Older generation histories are retained after recreation and must be purged separately if no longer needed.

Tests

Unit and orchestration-replay tests run with no external services. Live tests skip unless their environment is configured:

# Azure Blob write path against Azurite
AZURITE_CONNECTION_STRING="..." go test ./exporthistory/

# End-to-end against a live Durable Task Scheduler plus Azurite
DTS_EMULATOR_ENDPOINT="http://127.0.0.1:8080" DTS_TASK_HUB=default \
  AZURITE_CONNECTION_STRING="..." \
  go test ./tests/durabletaskscheduler/ -run TestDTSExportHistory

Documentation

Overview

Package exporthistory exports terminal orchestration histories to Azure Blob Storage using durable entities, orchestrations, and activities.

Maturity: preview

This package is a preview API. Its exported types, function signatures, the serialized shape of the ExportJob entity state, and the names of the system entity, orchestrators, and activities may change in a future release without a major version bump. Do not depend on it for production workloads that cannot tolerate a breaking change. Drain and delete export jobs before upgrading; legacy export-job state is not supported or migrated. Do not mix .NET or older Go export-history workers with these system tasks in the same task hub.

Model

An export job is a durable entity keyed by job ID. Creating a job transitions the entity to ExportJobStatusActive, mints a run token, and reserves an orchestration ID derived from the job ID and token before signaling Run. ExportJobDescription.OrchestratorInstanceID identifies that generation. Only the current generation can mutate job state; older orchestrations and already-scheduled activities can briefly remain in flight. The orchestration repeatedly lists terminal orchestration instances that match the job's filter, exports each instance's history, and commits a checkpoint back to the entity. A batch job completes when the task hub reports no further pages; an empty page with a continuation cursor advances the cursor and keeps going. A continuous job idles and lists again.

Recreation assigns a fresh orchestration ID and retains previous generations' orchestration histories. Delete atomically captures the current ID and removes entity state, then the client terminates and purges only that captured generation. A concurrent recreation is not affected by that cleanup. Every run and orchestration-originated mutation requires a matching nonempty token.

Usage

Workers register the system tasks and advertise them to the service:

registry := task.NewTaskRegistry()
store, err := exporthistory.NewAzureBlobHistoryStore(exporthistory.AzureBlobHistoryStoreOptions{
	ConnectionString: connectionString,
	ContainerName:    "history-exports",
})
if err != nil {
	return err
}
if err := exporthistory.Register(registry, exporthistory.WorkerOptions{
	Source: taskHubClient,
	Store:  store,
}); err != nil {
	return err
}
worker, err := durabletaskscheduler.NewWorker(
	options, registry, logger, exporthistory.WithExportHistory())

Clients create and inspect jobs:

exportClient, err := exporthistory.NewClient(taskHubClient, exporthistory.ClientOptions{
	ContainerName: "history-exports",
})
if err != nil {
	return err
}
job, err := exportClient.CreateJob(ctx, exporthistory.JobCreationOptions{
	Mode:              exporthistory.ExportModeBatch,
	CompletedTimeFrom: from,
	CompletedTimeTo:   to,
})
if err != nil {
	return err
}
description, err := job.Describe(ctx)

Limitations

Export requires a task hub that implements the instance-ID listing and history streaming management APIs. Extended sessions are not supported.

Index

Constants

View Source
const (
	// ExportJobEntityName is the system entity that owns export job state.
	ExportJobEntityName = "ExportJob"
	// ExportJobOrchestratorName is the system orchestrator that performs the export work.
	ExportJobOrchestratorName = "ExportJobOrchestrator"
	// ExecuteExportJobOperationOrchestratorName is the system orchestrator that
	// invokes a single export job entity operation on behalf of a client.
	ExecuteExportJobOperationOrchestratorName = "ExecuteExportJobOperationOrchestrator"
	// ListTerminalInstancesActivityName lists a page of terminal instance IDs.
	ListTerminalInstancesActivityName = "ListTerminalInstancesActivity"
	// ExportInstanceHistoryActivityName exports one instance's history.
	ExportInstanceHistoryActivityName = "ExportInstanceHistoryActivity"

	// OrchestratorInstanceIDPrefix prefixes each generation-specific export
	// orchestration ID. Read ExportJobDescription.OrchestratorInstanceID for
	// the current generation's ID.
	OrchestratorInstanceIDPrefix = "ExportJob-"
)
View Source
const (
	DefaultMaxInstancesPerBatch = 100
	// MaxInstancesPerBatchLimit is the inclusive upper bound accepted for
	// JobCreationOptions.MaxInstancesPerBatch.
	MaxInstancesPerBatchLimit = 1000
	// DefaultMaxParallelExports bounds concurrent per-instance export activities.
	DefaultMaxParallelExports = 32
	// DefaultSchemaVersion is the schema version stamped on exported objects.
	DefaultSchemaVersion = "1.0"
	// DefaultJobQueryPageSize is used when ExportJobQuery.PageSize is zero.
	DefaultJobQueryPageSize = 100
)

Defaults applied when a job creation option is not supplied.

View Source
const MaxCreationClockSkew = 5 * time.Minute

MaxCreationClockSkew is how far a batch job's CompletedTimeTo may sit ahead of the worker's clock and still be accepted by the ExportJob entity.

A client validates the upper bound strictly against its own clock, but the entity runs on a worker whose clock is independent. Without this tolerance a window the client accepted would be rejected by a worker running slightly behind, so the entity allows a bounded skew while clients stay strict.

Variables

View Source
var (
	// ErrJobNotFound identifies attempts to read a missing export job.
	ErrJobNotFound = errors.New("export job not found")
	// ErrJobInvalidTransition identifies an invalid export job state change.
	ErrJobInvalidTransition = errors.New("invalid export job state transition")
	// ErrValidation identifies invalid export job configuration or client input.
	ErrValidation = errors.New("invalid export job configuration")
	// ErrJobOperationFailed identifies a failed export job mutation.
	ErrJobOperationFailed = errors.New("export job operation failed")
)

Functions

func EntityID

func EntityID(jobID string) api.EntityID

EntityID returns the export job entity ID for jobID.

func ExecuteExportJobOperationOrchestrator

func ExecuteExportJobOperationOrchestrator(ctx *task.OrchestrationContext) (any, error)

ExecuteExportJobOperationOrchestrator invokes one export job entity operation. It must remain unversioned because export clients explicitly target the unversioned system orchestrator.

func ExportJobOrchestrator

func ExportJobOrchestrator(ctx *task.OrchestrationContext) (any, error)

ExportJobOrchestrator performs the export work for one job: it lists terminal instances matching the job's filter, exports each instance's history, and commits a checkpoint after every page.

It must remain unversioned because the entity starts it explicitly unversioned so it stays reachable when application default versioning is on.

func Register

func Register(registry *task.TaskRegistry, options WorkerOptions) error

Register adds the export history system entity, orchestrators, and activities to registry.

Every system task is registered unversioned so it stays reachable when an application enables default versioning; pair it with WithExportHistory so the worker's derived work-item filters keep accepting them under strict version matching.

func TerminalStatuses

func TerminalStatuses() []api.OrchestrationStatus

TerminalStatuses returns the orchestration runtime statuses an export job accepts, in the .NET declaration order.

func WithExportHistory

func WithExportHistory() durabletaskclient.TaskHubGrpcWorkerOption

WithExportHistory marks the export history system orchestrators and activities as unversioned so they remain routable when the worker runs with strict version matching, and so registry-derived work-item filters advertise them with the version they were registered under.

The export orchestrations are started unversioned, and an activity inherits its caller's version, so both kinds must be allow-listed. Use it together with Register; on its own it changes no behavior.

Types

type AzureBlobHistoryStore

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

AzureBlobHistoryStore writes exported history objects to Azure Blob Storage.

func NewAzureBlobHistoryStore

func NewAzureBlobHistoryStore(options AzureBlobHistoryStoreOptions) (*AzureBlobHistoryStore, error)

NewAzureBlobHistoryStore constructs a production Azure Blob Storage export store.

func (*AzureBlobHistoryStore) DefaultContainer

func (s *AzureBlobHistoryStore) DefaultContainer() string

DefaultContainer returns the container configured for this store.

func (*AzureBlobHistoryStore) Write

func (s *AzureBlobHistoryStore) Write(ctx context.Context, object ExportObject) error

Write uploads object, creating its container on first use.

type AzureBlobHistoryStoreOptions

type AzureBlobHistoryStoreOptions struct {
	ConnectionString string
	AccountURL       string
	Credential       azcore.TokenCredential
	// ContainerName is the default destination container. It is required so a
	// misconfigured worker fails at construction instead of at export time.
	ContainerName string
	// AllowInsecureHTTP permits plaintext endpoints such as Azurite.
	AllowInsecureHTTP bool
	// AllowedContainers optionally restricts the containers an export job may
	// write to. When empty only ContainerName is allowed.
	AllowedContainers []string
	// AllowAnyContainer disables container allow-listing entirely.
	AllowAnyContainer bool
}

AzureBlobHistoryStoreOptions configures AzureBlobHistoryStore. Exactly one authentication mode must be supplied: ConnectionString, or AccountURL with Credential.

type Client

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

Client creates and inspects export jobs in a task hub.

func NewClient

func NewClient(client *durabletaskclient.TaskHubGrpcClient, options ClientOptions) (*Client, error)

NewClient returns an export history client backed by client.

func (*Client) CreateJob

func (c *Client) CreateJob(ctx context.Context, options JobCreationOptions) (*JobClient, error)

CreateJob normalizes options, resolves the destination, creates the job, and returns a handle for it.

func (*Client) GetJob

func (c *Client) GetJob(ctx context.Context, jobID string) (*ExportJobDescription, error)

GetJob returns the current description of jobID, or a NotFoundError when the job does not exist.

func (*Client) JobClient

func (c *Client) JobClient(jobID string) (*JobClient, error)

JobClient returns a handle for jobID without contacting the service.

func (*Client) ListJobs

func (c *Client) ListJobs(ctx context.Context, query ExportJobQuery) (*ExportJobQueryResult, error)

ListJobs returns one page of export jobs matching query. Status and creation-time filters are applied after the service returns the entity page, so a page can contain fewer than PageSize items and an empty page can still carry a continuation token.

type ClientOptions

type ClientOptions struct {
	// ContainerName is the default destination container.
	ContainerName string
	// Prefix is the default destination path prefix. When empty each job gets
	// the mode-and-ID prefix "<mode>-<jobId>/", matching .NET.
	Prefix string
}

ClientOptions configures Client. ContainerName and Prefix supply the destination for jobs created without an explicit ExportDestination.

type CommitCheckpointRequest

type CommitCheckpointRequest struct {
	ScannedInstances  int64             `json:"ScannedInstances"`
	ExportedInstances int64             `json:"ExportedInstances"`
	Checkpoint        *ExportCheckpoint `json:"Checkpoint,omitempty"`
	Failures          []ExportFailure   `json:"Failures,omitempty"`
	// RunToken fences the commit to the run generation that produced it. See
	// [ExportJobState.RunToken].
	RunToken string `json:"RunToken,omitempty"`
}

CommitCheckpointRequest is the entity input that records a batch's progress. A nil Checkpoint keeps the current cursor so the same batch can be retried; combined with a non-empty Failures list it implicitly fails the job.

type ExportCheckpoint

type ExportCheckpoint struct {
	LastInstanceKey string `json:"LastInstanceKey,omitempty"`
}

ExportCheckpoint records the durable pagination cursor of an export job.

type ExportDestination

type ExportDestination struct {
	Container string `json:"Container"`
	Prefix    string `json:"Prefix,omitempty"`
}

ExportDestination is the blob container and optional path prefix that receives exported objects.

func (ExportDestination) BlobPath

func (d ExportDestination) BlobPath(name string) string

BlobPath joins the destination prefix with name using a single separator.

func (ExportDestination) Validate

func (d ExportDestination) Validate() error

Validate reports whether the destination can receive exported objects.

type ExportFailure

type ExportFailure struct {
	InstanceID   string    `json:"InstanceId"`
	Reason       string    `json:"Reason"`
	AttemptCount int       `json:"AttemptCount"`
	LastAttempt  time.Time `json:"LastAttempt"`
}

ExportFailure describes one instance that could not be exported.

type ExportFilter

type ExportFilter struct {
	CompletedTimeFrom time.Time
	CompletedTimeTo   time.Time
	RuntimeStatus     []api.OrchestrationStatus
}

ExportFilter selects the orchestration instances an export job reads. Only terminal runtime statuses are supported. A zero CompletedTimeTo means the window has no upper bound.

func (ExportFilter) MarshalJSON

func (f ExportFilter) MarshalJSON() ([]byte, error)

func (*ExportFilter) UnmarshalJSON

func (f *ExportFilter) UnmarshalJSON(data []byte) error

type ExportFormat

type ExportFormat struct {
	Kind          ExportFormatKind `json:"Kind"`
	SchemaVersion string           `json:"SchemaVersion"`
}

ExportFormat describes the serialization and schema of exported objects.

func DefaultExportFormat

func DefaultExportFormat() ExportFormat

DefaultExportFormat is gzip-compressed JSONL at the default schema version.

func (ExportFormat) ContentType

func (f ExportFormat) ContentType() string

ContentType returns the MIME type exported objects are stored under.

JSONL objects are gzip-compressed and stored as opaque gzip files, so they use application/gzip and carry no content coding: the bytes a reader downloads are always the gzip stream the object name promises.

func (ExportFormat) FileExtension

func (f ExportFormat) FileExtension() string

FileExtension returns the object suffix used for this format.

type ExportFormatKind

type ExportFormatKind int

ExportFormatKind selects the serialization of an exported history.

const (
	// ExportFormatJSONL writes one history event per line and gzip-compresses the object.
	ExportFormatJSONL ExportFormatKind = iota
	// ExportFormatJSON writes an uncompressed JSON array of history events.
	ExportFormatJSON
)

func (ExportFormatKind) IsValid

func (k ExportFormatKind) IsValid() bool

IsValid reports whether k is a supported format kind.

func (ExportFormatKind) MarshalJSON

func (k ExportFormatKind) MarshalJSON() ([]byte, error)

MarshalJSON writes the .NET-compatible string form of the format kind.

func (ExportFormatKind) String

func (k ExportFormatKind) String() string

func (*ExportFormatKind) UnmarshalJSON

func (k *ExportFormatKind) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts either the .NET string form or the numeric form.

type ExportJobConfiguration

type ExportJobConfiguration struct {
	Mode                 ExportMode        `json:"Mode"`
	Filter               ExportFilter      `json:"Filter"`
	Destination          ExportDestination `json:"Destination"`
	Format               ExportFormat      `json:"Format"`
	MaxParallelExports   int               `json:"MaxParallelExports"`
	MaxInstancesPerBatch int               `json:"MaxInstancesPerBatch"`
}

ExportJobConfiguration is the durable configuration of an export job.

type ExportJobDescription

type ExportJobDescription struct {
	JobID          string
	Status         ExportJobStatus
	CreatedAt      time.Time
	LastModifiedAt time.Time
	Config         *ExportJobConfiguration
	// OrchestratorInstanceID identifies the current generation, reserved during
	// Create before its Run signal is delivered. Recreation assigns a new ID.
	OrchestratorInstanceID string
	// ScannedInstances includes entries counted again when a page is re-scanned.
	ScannedInstances int64
	// ExportedInstances includes re-exports, matching ExportJobState's processing-total semantics.
	ExportedInstances  int64
	LastError          string
	Checkpoint         *ExportCheckpoint
	LastCheckpointTime time.Time
}

ExportJobDescription is the client-facing view of an export job.

type ExportJobOperationRequest

type ExportJobOperationRequest struct {
	EntityID      api.EntityID `json:"EntityId"`
	OperationName string       `json:"OperationName"`
	Input         any          `json:"Input,omitempty"`
}

ExportJobOperationRequest is the input consumed by ExecuteExportJobOperationOrchestrator.

type ExportJobQuery

type ExportJobQuery struct {
	Status *ExportJobStatus
	// JobIDPrefix restricts results to jobs whose ID starts with this value.
	JobIDPrefix string
	// CreatedFrom is an exclusive lower creation-time bound, matching .NET.
	CreatedFrom time.Time
	// CreatedTo is an exclusive upper creation-time bound, matching .NET.
	CreatedTo         time.Time
	PageSize          int32
	ContinuationToken string
}

ExportJobQuery filters a single page of export jobs. Status and creation-time filters are applied after the service returns the entity page, so a page can contain fewer than PageSize items.

type ExportJobQueryResult

type ExportJobQueryResult struct {
	Jobs              []*ExportJobDescription
	ContinuationToken string
}

ExportJobQueryResult is one page of export jobs.

type ExportJobRunRequest

type ExportJobRunRequest struct {
	JobEntityID     api.EntityID `json:"JobEntityId"`
	ProcessedCycles int          `json:"ProcessedCycles"`
	// RunToken is the job generation this orchestration run belongs to. Every
	// entity mutation the run performs carries it, so a run left over from a
	// deleted-and-recreated job cannot alter the new one. See
	// [ExportJobState.RunToken].
	RunToken string `json:"RunToken,omitempty"`
	// ContinuedExecution marks an execution that resumed through ContinueAsNew
	// rather than a fresh start. Its zero value identifies the first execution,
	// which is the only one that treats a missing job as an error rather than
	// as a job that was deleted while it ran.
	ContinuedExecution bool `json:"ContinuedExecution,omitempty"`
}

ExportJobRunRequest is the input consumed by ExportJobOrchestrator.

type ExportJobState

type ExportJobState struct {
	Status             ExportJobStatus         `json:"Status"`
	Config             *ExportJobConfiguration `json:"Config,omitempty"`
	Checkpoint         *ExportCheckpoint       `json:"Checkpoint,omitempty"`
	CreatedAt          *time.Time              `json:"CreatedAt,omitempty"`
	LastModifiedAt     *time.Time              `json:"LastModifiedAt,omitempty"`
	LastCheckpointTime *time.Time              `json:"LastCheckpointTime,omitempty"`
	LastError          string                  `json:"LastError,omitempty"`
	// ScannedInstances is cumulative scan progress, including repeated pages.
	ScannedInstances int64 `json:"ScannedInstances"`
	// ExportedInstances counts successful exports, including re-exports; it is not a distinct-execution count.
	ExportedInstances      int64  `json:"ExportedInstances"`
	OrchestratorInstanceID string `json:"OrchestratorInstanceId,omitempty"`
	// RunToken identifies the job's current run generation. Every Create mints
	// a new one, so a run that started before the job was deleted and recreated
	// carries a stale token and is fenced out of the new generation's state.
	// Mutations require a matching, nonempty token.
	RunToken string `json:"RunToken,omitempty"`
}

ExportJobState is the persisted entity state of an export job.

type ExportJobStatus

type ExportJobStatus int

ExportJobStatus is the persisted lifecycle status of an export job.

const (
	// ExportJobStatusPending is the zero value: the entity has no configuration yet.
	ExportJobStatusPending ExportJobStatus = iota
	// ExportJobStatusActive means the job is running.
	ExportJobStatusActive
	// ExportJobStatusFailed means the job stopped because of an error.
	ExportJobStatusFailed
	// ExportJobStatusCompleted means the job exported its whole window.
	ExportJobStatusCompleted
)

func (ExportJobStatus) IsValid

func (s ExportJobStatus) IsValid() bool

IsValid reports whether s is a persisted export job status.

func (ExportJobStatus) String

func (s ExportJobStatus) String() string

type ExportMode

type ExportMode int

ExportMode selects how an export job consumes its filter window.

const (
	// ExportModeBatch exports a fixed completion-time window and then completes.
	ExportModeBatch ExportMode = 1
	// ExportModeContinuous tails terminal instances indefinitely.
	ExportModeContinuous ExportMode = 2
)

func (ExportMode) IsValid

func (m ExportMode) IsValid() bool

IsValid reports whether m is a mode the export job supports.

func (ExportMode) String

func (m ExportMode) String() string

type ExportObject

type ExportObject struct {
	// Container is the destination container name.
	Container string
	// Name is the full object path inside the container, prefix included.
	Name string
	// Content is the object body, already compressed when the format requires it.
	Content io.Reader
	// ContentType is the MIME type of Content. A compressed format reports the
	// compressed type, such as application/gzip, rather than declaring the
	// compression as a separate content coding, so a reader always receives the
	// exact bytes the object name describes.
	ContentType string
	// Metadata is attached to the stored object.
	Metadata map[string]string
}

ExportObject is one serialized orchestration history ready to be persisted.

type ExportRequest

type ExportRequest struct {
	InstanceID  string            `json:"InstanceId"`
	Destination ExportDestination `json:"Destination"`
	Format      ExportFormat      `json:"Format"`
}

ExportRequest is the input consumed by ExportInstanceHistoryActivity.

type ExportResult

type ExportResult struct {
	InstanceID  string `json:"InstanceId"`
	ExecutionID string `json:"ExecutionId,omitempty"`
	Success     bool   `json:"Success"`
	Error       string `json:"Error,omitempty"`
	BlobPath    string `json:"BlobPath,omitempty"`
	EventCount  int    `json:"EventCount,omitempty"`
}

ExportResult reports the outcome of exporting one instance. Per-instance failures are collected rather than thrown so a batch can report every failing instance at once.

type HistorySource

type HistorySource interface {
	// ListInstanceIDs returns one page of instance IDs matching the query.
	ListInstanceIDs(ctx context.Context, query api.InstanceIDQuery) (*api.InstanceIDQueryResult, error)
	// FetchOrchestrationMetadata returns the instance's current metadata.
	FetchOrchestrationMetadata(
		ctx context.Context,
		id api.InstanceID,
		opts ...api.FetchOrchestrationMetadataOptions,
	) (*api.OrchestrationMetadata, error)
	// StreamOrchestrationHistory delivers events serially, stopping when the
	// handler returns an error or ctx is canceled.
	StreamOrchestrationHistory(
		ctx context.Context,
		id api.InstanceID,
		query api.HistoryQuery,
		handler api.HistoryEventHandler,
	) error
}

HistorySource supplies the management reads an export job needs. It is implemented by github.com/microsoft/durabletask-go/client.TaskHubGrpcClient and by the Durable Task Scheduler client.

type InstancePage

type InstancePage struct {
	InstanceIDs    []string          `json:"InstanceIds"`
	NextCheckpoint *ExportCheckpoint `json:"NextCheckpoint,omitempty"`
}

InstancePage is one page of terminal instance IDs plus the next cursor.

type InvalidTransitionError

type InvalidTransitionError struct {
	JobID     string
	From      ExportJobStatus
	To        ExportJobStatus
	Operation string
}

InvalidTransitionError reports a lifecycle transition the job does not allow.

func (*InvalidTransitionError) DurableTaskErrorProperties

func (e *InvalidTransitionError) DurableTaskErrorProperties() map[string]any

func (*InvalidTransitionError) DurableTaskErrorType

func (*InvalidTransitionError) DurableTaskErrorType() api.ErrorType

func (*InvalidTransitionError) Error

func (e *InvalidTransitionError) Error() string

func (*InvalidTransitionError) Unwrap

func (e *InvalidTransitionError) Unwrap() error

type JobClient

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

JobClient manages one export job.

func (*JobClient) Create

func (c *JobClient) Create(ctx context.Context, options JobCreationOptions) error

Create creates the job, or recreates it in place when it previously failed or completed. Creating a job that is already Active fails with an InvalidTransitionError.

Each successful Create reserves a new generation-specific orchestration ID. Previous generations' orchestration histories are not terminated or purged.

func (*JobClient) Delete

func (c *JobClient) Delete(ctx context.Context) error

Delete removes the job's entity state and stops the generation it removed.

The entity delete and the orchestration cleanup are not atomic: the entity is deleted first and returns its captured orchestration ID. Cleanup targets only that ID, so a concurrent Create cannot have its new generation terminated or purged by this Delete. A missing job or orchestration is not an error.

func (*JobClient) Describe

func (c *JobClient) Describe(ctx context.Context) (*ExportJobDescription, error)

Describe returns the job's current description, or a NotFoundError.

func (*JobClient) ID

func (c *JobClient) ID() string

ID returns the job identifier managed by this handle.

type JobCreationOptions

type JobCreationOptions struct {
	JobID                string
	Mode                 ExportMode
	CompletedTimeFrom    time.Time
	CompletedTimeTo      time.Time
	Destination          *ExportDestination
	Format               *ExportFormat
	RuntimeStatus        []api.OrchestrationStatus
	MaxInstancesPerBatch int
}

JobCreationOptions configures a new export job.

Zero values select defaults: an empty JobID generates one, a nil Format uses DefaultExportFormat, an empty RuntimeStatus exports every terminal status, and a zero MaxInstancesPerBatch uses DefaultMaxInstancesPerBatch. A zero CompletedTimeTo means the window has no upper bound, which is required for ExportModeContinuous and rejected for ExportModeBatch.

Destination may be left nil when the client is configured with a default container; the client fills it in before the job reaches the entity.

func (JobCreationOptions) MarshalJSON

func (o JobCreationOptions) MarshalJSON() ([]byte, error)

func (JobCreationOptions) Normalize

func (o JobCreationOptions) Normalize() (JobCreationOptions, error)

Normalize validates the options against the wall clock and returns a copy with every default applied. It generates a job ID when one is not supplied, so calling it twice produces two different jobs.

func (*JobCreationOptions) UnmarshalJSON

func (o *JobCreationOptions) UnmarshalJSON(data []byte) error

func (JobCreationOptions) Validate

func (o JobCreationOptions) Validate() error

Validate reports whether the options are usable, applying the same rules as JobCreationOptions.Normalize without mutating anything.

type ListTerminalInstancesRequest

type ListTerminalInstancesRequest struct {
	CompletedTimeFrom    time.Time                 `json:"CompletedTimeFrom"`
	CompletedTimeTo      *time.Time                `json:"CompletedTimeTo"`
	RuntimeStatus        []api.OrchestrationStatus `json:"RuntimeStatus,omitempty"`
	LastInstanceKey      string                    `json:"LastInstanceKey,omitempty"`
	MaxInstancesPerBatch int                       `json:"MaxInstancesPerBatch"`
}

ListTerminalInstancesRequest is the input consumed by ListTerminalInstancesActivity. A zero CompletedTimeTo means the window has no upper bound.

type MarkAsCompletedRequest

type MarkAsCompletedRequest struct {
	// RunToken fences the completion to the run generation that produced it.
	// See [ExportJobState.RunToken].
	RunToken string `json:"RunToken,omitempty"`
}

MarkAsCompletedRequest is the entity input of the MarkAsCompleted operation.

type MarkAsFailedRequest

type MarkAsFailedRequest struct {
	// RunToken fences the failure to the run generation that produced it. See
	// [ExportJobState.RunToken].
	RunToken string `json:"RunToken,omitempty"`
	// Error is the message recorded on the job. It is optional.
	Error string `json:"Error,omitempty"`
}

MarkAsFailedRequest is the entity input of the MarkAsFailed operation.

type NotFoundError

type NotFoundError struct{ JobID string }

NotFoundError reports an export job that does not exist.

func (*NotFoundError) DurableTaskErrorProperties

func (e *NotFoundError) DurableTaskErrorProperties() map[string]any

func (*NotFoundError) DurableTaskErrorType

func (*NotFoundError) DurableTaskErrorType() api.ErrorType

func (*NotFoundError) Error

func (e *NotFoundError) Error() string

func (*NotFoundError) Unwrap

func (e *NotFoundError) Unwrap() error

type OperationError

type OperationError struct {
	JobID          string
	Operation      string
	RuntimeStatus  api.OrchestrationStatus
	FailureDetails *api.FailureDetails
}

OperationError reports an export job mutation orchestration that did not complete successfully and whose failure could not be mapped to a more specific typed error.

func (*OperationError) DurableTaskErrorType

func (*OperationError) DurableTaskErrorType() api.ErrorType

func (*OperationError) Error

func (e *OperationError) Error() string

func (*OperationError) Unwrap

func (e *OperationError) Unwrap() error

type RunJobRequest

type RunJobRequest struct {
	// RunToken fences the run signal to the generation that emitted it. See
	// [ExportJobState.RunToken].
	RunToken string `json:"RunToken,omitempty"`
}

RunJobRequest is the entity input of the Run operation, which starts the job's export orchestration.

type Store

type Store interface {
	Write(ctx context.Context, object ExportObject) error
}

Store persists exported history objects. It is deliberately narrower than github.com/microsoft/durabletask-go/payload.AzureBlobStore, whose large-payload contract assigns random object names and owns a single container, because export jobs choose deterministic names in a caller-chosen container.

Implementations must be safe for concurrent use and should treat a repeated write of the same object name as an overwrite, since an activity retry can re-export an instance whose object already exists. Write must consume Content synchronously, respect ctx, and publish only after a clean EOF. A read error must abort the write, leaving an existing object unchanged. Content is single-use; retrying an export opens a fresh history stream rather than replaying this reader.

type ValidationError

type ValidationError struct {
	JobID   string
	Message string
}

ValidationError reports invalid export job options or configuration.

func (*ValidationError) DurableTaskErrorProperties

func (e *ValidationError) DurableTaskErrorProperties() map[string]any

func (*ValidationError) DurableTaskErrorType

func (*ValidationError) DurableTaskErrorType() api.ErrorType

func (*ValidationError) Error

func (e *ValidationError) Error() string

func (*ValidationError) Unwrap

func (e *ValidationError) Unwrap() error

type WorkerOptions

type WorkerOptions struct {
	// Source supplies the management reads the export activities perform. It is
	// required and is normally the same task hub client the worker connects with.
	Source HistorySource
	// Store persists exported objects. It is required.
	Store Store
	// HistoryQuery bounds each instance-history read. Zero values select the
	// api package defaults.
	HistoryQuery api.HistoryQuery
}

WorkerOptions configures the export history system tasks a worker runs.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL