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
- Variables
- func EntityID(jobID string) api.EntityID
- func ExecuteExportJobOperationOrchestrator(ctx *task.OrchestrationContext) (any, error)
- func ExportJobOrchestrator(ctx *task.OrchestrationContext) (any, error)
- func Register(registry *task.TaskRegistry, options WorkerOptions) error
- func TerminalStatuses() []api.OrchestrationStatus
- func WithExportHistory() durabletaskclient.TaskHubGrpcWorkerOption
- type AzureBlobHistoryStore
- type AzureBlobHistoryStoreOptions
- type Client
- func (c *Client) CreateJob(ctx context.Context, options JobCreationOptions) (*JobClient, error)
- func (c *Client) GetJob(ctx context.Context, jobID string) (*ExportJobDescription, error)
- func (c *Client) JobClient(jobID string) (*JobClient, error)
- func (c *Client) ListJobs(ctx context.Context, query ExportJobQuery) (*ExportJobQueryResult, error)
- type ClientOptions
- type CommitCheckpointRequest
- type ExportCheckpoint
- type ExportDestination
- type ExportFailure
- type ExportFilter
- type ExportFormat
- type ExportFormatKind
- type ExportJobConfiguration
- type ExportJobDescription
- type ExportJobOperationRequest
- type ExportJobQuery
- type ExportJobQueryResult
- type ExportJobRunRequest
- type ExportJobState
- type ExportJobStatus
- type ExportMode
- type ExportObject
- type ExportRequest
- type ExportResult
- type HistorySource
- type InstancePage
- type InvalidTransitionError
- type JobClient
- type JobCreationOptions
- type ListTerminalInstancesRequest
- type MarkAsCompletedRequest
- type MarkAsFailedRequest
- type NotFoundError
- type OperationError
- type RunJobRequest
- type Store
- type ValidationError
- type WorkerOptions
Constants ¶
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-" )
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.
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 ¶
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 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 ¶
CreateJob normalizes options, resolves the destination, creates the job, and returns a handle for it.
func (*Client) GetJob ¶
GetJob returns the current description of jobID, or a NotFoundError when the job does not exist.
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 ¶
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.
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 ¶
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.