Help us shape the future of Go! Take the 2026 Go Developer Survey

observation

package
v0.9.1 Latest Latest
Warning

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

 Go to latest Published: Oct 10, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package observation administers observers and waits for processing of committed append coordinates. Acceptance, processing and durable checkpoint evidence differ.

Index

Examples

Constants

View Source 
const ClientGrace = 200 * time.Millisecond

ClientGrace gives the server 200ms to report outstanding observers after its budget.

View Source 
const DefaultTimeout = 5 * time.Second

DefaultTimeout is the five-second server budget used by the C# client.

View Source 
const InfiniteTimeout time.Duration = -1

InfiniteTimeout disables server and helper deadlines, but not caller cancellation.

Variables

This section is empty.

Functions

This section is empty.

Types

type CannotWaitError 

type CannotWaitError struct {
	// Store is the original append store.
	Store metadata.StoreName
	// Sequence is the original append sequence.
	Sequence events.SequenceID
}

CannotWaitError reports committed work without an observer administration surface. An unavailable append tail is different and completes trivially.

func (*CannotWaitError) Error 

func (e *CannotWaitError) Error() string

type Completion 

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

Completion owns immutable original append coordinates and per-event-type tails. Zero means no committed tail and completes trivially, not proof of a commit. Construct from an append operation's Completion method or NewCompletion.

func NewCompletion 

func NewCompletion(store metadata.StoreName, namespace metadata.Namespace, sequence events.SequenceID, types []events.TypeRef, positions []events.SequenceNumber) (Completion, error)

NewCompletion snapshots exact input types zipped with confirmed positions. Repeated type IDs use their greatest actual position and that entry's generation, never the whole batch tail. Position zero is valid. Empty arrays or one Unavailable position mean no committed tail. Unknown append outcomes must not use this constructor.

func (Completion) Target 

func (c Completion) Target() CompletionTarget

Target returns a detached legacy coordinate snapshot. Exact generations remain private in Completion; use this Completion, not a reconstructed catalog guess, to wait.

type CompletionResult 

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

CompletionResult is immutable server processing evidence, not a durable sink checkpoint or transaction guarantee. Zero is not success.

func (CompletionResult) FailedPartitions 

func (r CompletionResult) FailedPartitions() []FailedPartition

FailedPartitions returns owned immutable failure snapshots.

func (CompletionResult) IsSuccess 

func (r CompletionResult) IsSuccess() bool

IsSuccess reports the server's processing success (or a trivial unavailable tail).

func (CompletionResult) OutstandingObservers 

func (r CompletionResult) OutstandingObservers() []ID

OutstandingObservers returns an owned list of observers outstanding at timeout.

func (CompletionResult) TimedOut 

func (r CompletionResult) TimedOut() bool

TimedOut reports the server budget or client grace deadline expiring.

func (CompletionResult) Trivial 

func (r CompletionResult) Trivial() bool

Trivial distinguishes no committed tail from actual observer-processing evidence.

type CompletionTarget 

type CompletionTarget struct {
	// Store identifies the event store.
	Store metadata.StoreName
	// Namespace identifies the isolated namespace.
	Namespace metadata.Namespace
	// Sequence identifies the event sequence.
	Sequence events.SequenceID
	// First is the first committed position; nil means no committed work.
	First *events.SequenceNumber
	// EventTypeTails maps each appended event type to its last committed position.
	EventTypeTails map[events.TypeID]events.SequenceNumber
}

CompletionTarget identifies committed positions without retaining a client. Zero-valued targets do not represent committed work.

type FailedAttempt 

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

FailedAttempt is an immutable failure attempt diagnostic.

func (FailedAttempt) Kind 

func (a FailedAttempt) Kind() FailureKind

Kind returns the failure classification.

func (FailedAttempt) Messages 

func (a FailedAttempt) Messages() []string

Messages returns owned diagnostic messages; they may contain sensitive data.

func (FailedAttempt) Occurred 

func (a FailedAttempt) Occurred() time.Time

Occurred returns the failure instant with its original offset.

func (FailedAttempt) Position 

func (a FailedAttempt) Position() events.SequenceNumber

Position returns the failed event's sequence position.

func (FailedAttempt) StackTrace 

func (a FailedAttempt) StackTrace() string

StackTrace returns the recorded stack, which may contain sensitive data.

type FailedPartition 

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

FailedPartition is an immutable diagnostic snapshot. Resolved and quarantined remain independent; a cleared quarantine does not establish successful recovery.

func (FailedPartition) Attempts 

func (p FailedPartition) Attempts() []FailedAttempt

Attempts returns an owned slice of immutable attempts in server order.

func (FailedPartition) ID 

func (p FailedPartition) ID() uuid.UUID

ID returns the failure record's UUID.

func (FailedPartition) IsQuarantined 

func (p FailedPartition) IsQuarantined() bool

IsQuarantined reports the kernel quarantine flag.

func (FailedPartition) IsResolved 

func (p FailedPartition) IsResolved() bool

IsResolved reports the kernel resolution flag.

func (FailedPartition) Observer 

func (p FailedPartition) Observer() ID

Observer returns the failed observer identity.

func (FailedPartition) Partition 

func (p FailedPartition) Partition() Partition

Partition returns the failed partition key.

type FailureKind 

type FailureKind int32

FailureKind distinguishes failures, preserving unknown numeric values. 

const (
	// UnknownFailure means the kernel supplied no classification.
	UnknownFailure FailureKind = iota
	// HandlingFailure means the event handler failed.
	HandlingFailure
	// TimeoutFailure means handling timed out.
	TimeoutFailure
	// DisconnectedFailure means the client disconnected.
	DisconnectedFailure
)

type ID 

type ID string

ID is a stable observer identifier shared by projections, reactors and reducers.

type Information 

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

Information is an immutable server snapshot. Zero is invalid. A running state or a tail is not a guarantee of durable checkpoint persistence.

func (Information) EventTypes 

func (i Information) EventTypes() []events.TypeRef

EventTypes returns owned exact subscribed event references in server order.

func (Information) HandledEventCount 

func (i Information) HandledEventCount() uint64

HandledEventCount returns the reported count.

func (Information) ID 

func (i Information) ID() ID

ID returns the observer identity.

func (Information) IsReplayable 

func (i Information) IsReplayable() bool

IsReplayable reports the server's replay policy.

func (Information) IsSubscribed 

func (i Information) IsSubscribed() bool

IsSubscribed reports a server-side subscription, not local goroutine lifetime. False when SubscriptionKnown is false means unavailable, NOT disconnected.

func (Information) LastHandled 

func (i Information) LastHandled() events.SequenceNumber

LastHandled returns the last handled position, preserving Unavailable.

func (Information) Next 

func (i Information) Next() events.SequenceNumber

Next returns the reported next position, preserving reserved sentinels.

func (Information) Owner 

func (i Information) Owner() Owner

Owner returns the execution owner.

func (Information) Remove 

func (i Information) Remove(ctx context.Context) (RemovalResult, error)

Remove requests store-wide removal using this snapshot's identity and sequence. The current definition is checked again; a stale sequence fails before mutation.

func (Information) RunningState 

func (i Information) RunningState() RunningState

RunningState returns runtime state independently of progress.

func (Information) Sequence 

func (i Information) Sequence() events.SequenceID

Sequence returns the source event sequence.

func (Information) SubscriptionKnown 

func (i Information) SubscriptionKnown() bool

SubscriptionKnown reports whether subscription state was queried. Only Get supplies it; List does not ask the kernel's live observer activation.

func (Information) Tail 

func (i Information) Tail() events.SequenceNumber

Tail returns the observer's reported source tail, preserving Unavailable.

func (Information) Type 

func (i Information) Type() Type

Type returns the observer kind.

type OutcomeUnknownError 

type OutcomeUnknownError = administration.OutcomeUnknownError

OutcomeUnknownError means an administration mutation may have taken effect despite an RPC/protocol failure. Do not automatically retry it.

type Owner 

type Owner int32

Owner identifies where an observer runs, preserving unknown numeric values. 

const (
	// NoOwner means no owner is reported.
	NoOwner Owner = iota
	// ClientOwner means a connected client owns execution.
	ClientOwner
	// KernelOwner means the kernel owns execution.
	KernelOwner
)

type Partition 

type Partition string

Partition is the opaque observer partition key, not necessarily an event source.

type QuarantineOutcome 

type QuarantineOutcome int32

QuarantineOutcome describes whether a partition fence was cleared. 

const (
	// QuarantineCleared means the partition quarantine was cleared.
	QuarantineCleared QuarantineOutcome = iota
	// QuarantineNotFound means the partition record was not found.
	QuarantineNotFound
	// NotQuarantined means the partition was not quarantined.
	NotQuarantined
)

type QuarantineResult 

type QuarantineResult struct {
	// Outcome is the quarantine-clearing result.
	Outcome QuarantineOutcome
	// RetryRequested records the caller's explicit retry instruction.
	RetryRequested bool
	// RetryOutcome is meaningful only when RetryRequested and Outcome is QuarantineCleared.
	RetryOutcome RecoveryOutcome
}

QuarantineResult is an owned value. RetryRequested qualifies RetryOutcome; when false, the proto3 default Started value is NOT recovery evidence.

type RecoveryOutcome 

type RecoveryOutcome int32

RecoveryOutcome preserves retry acceptance versus recovery refusals. 

const (
	// RecoveryStarted means recovery was accepted, not completed.
	RecoveryStarted RecoveryOutcome = iota
	// PartitionNotFound means no failed partition was found.
	PartitionNotFound
	// ObserverQuarantined means the observer fence prevented recovery.
	ObserverQuarantined
	// PartitionQuarantined means the partition fence prevented recovery.
	PartitionQuarantined
)

type RemovalOutcome 

type RemovalOutcome int32

RemovalOutcome preserves all server refusal distinctions. 

const (
	// Removed means the server removed the persistent observer records.
	Removed RemovalOutcome = iota
	// ObserverNotFound means no observer was found to remove.
	ObserverNotFound
	// ObserverActive means execution in some namespace prevents removal.
	ObserverActive
	// ObserverSubscribed means a subscribed client prevents removal.
	ObserverSubscribed
)

type RemovalResult 

type RemovalResult struct {
	// Outcome is the exact server result.
	Outcome RemovalOutcome
	// BlockingNamespace identifies the namespace preventing removal, when supplied.
	BlockingNamespace metadata.Namespace
}

RemovalResult is an owned value describing persistent removal, not local unsubscribe.

type RunningState 

type RunningState int32

RunningState describes runtime state, preserving unknown numeric values. 

const (
	// UnknownState means no runtime state is reported.
	UnknownState RunningState = iota
	// Active means the observer is active, not necessarily caught up.
	Active
	// Suspended means observation is suspended.
	Suspended
	// Replaying means replay is underway.
	Replaying
	// Disconnected means no active connection.
	Disconnected
	// Quarantined means automatic execution is fenced.
	Quarantined
)

type Service 

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

Service administers observers in an immutable store/namespace. It borrows the transport and starts no background work. Calls are concurrent-safe, context-first and never retry mutations. Remove is deliberately store-wide; see its contract.

func New 

func New(store metadata.StoreName, namespace metadata.Namespace, conn grpc.ClientConnInterface) (*Service, error)

New constructs a low-level service without I/O. Prefer EventStore.Observers for connection/registration barriers. The caller owns the connection.

func (*Service) ClearFailedPartitions 

func (s *Service) ClearFailedPartitions(ctx context.Context, id ID, sequence events.SequenceID) error

ClearFailedPartitions explicitly discards failure records, not the underlying cause. It does not claim recovery or successful observer processing.

func (*Service) ClearObserverQuarantine 

func (s *Service) ClearObserverQuarantine(ctx context.Context, id ID, sequence events.SequenceID) error

ClearObserverQuarantine explicitly clears the observer-level fence. The empty response acknowledges the command only; it cannot prove processing completion.

func (*Service) ClearPartitionQuarantine 

func (s *Service) ClearPartitionQuarantine(ctx context.Context, id ID, sequence events.SequenceID, partition Partition, retryImmediately bool) (QuarantineResult, error)

ClearPartitionQuarantine explicitly clears a fence, optionally requesting immediate recovery in the same call. Clearing does not prove recovery completed.

func (*Service) FailedPartitions 

func (s *Service) FailedPartitions(ctx context.Context, observer ID) ([]FailedPartition, error)

FailedPartitions returns all failure records when observer is empty, or those for that observer. Attempts, resolution and quarantine flags remain distinct.

func (*Service) Get 

func (s *Service) Get(ctx context.Context, id ID, sequence events.SequenceID) (*Information, error)

Get returns one observer's state. gRPC NotFound is (nil,nil); a malformed or mismatched successful reply is an error, not fabricated absence.

func (*Service) List 

func (s *Service) List(ctx context.Context) ([]Information, error)

List returns immutable observer snapshots in the current namespace. Subscription state is unavailable here; use Get for an authoritative subscription query.

func (*Service) Remove 

func (s *Service) Remove(ctx context.Context, id ID) (RemovalResult, error)

Remove deletes persistent observer records across ALL namespaces in this store. It first resolves the sequence from List's stored definition, never defaulting to EventLog. Missing, ambiguous or unsupported definition reads fail closed. The kernel refuses active/subscribed observers and leaves read-model data and sink containers untouched. This does not unregister local declarations or fence concurrent re-registration; stop declaring applications before removal.

func (*Service) RemoveFrom 

func (s *Service) RemoveFrom(ctx context.Context, id ID, sequence events.SequenceID) (RemovalResult, error)

RemoveFrom requests removal for an explicit sequence, checked against the current stored definition before mutation. A wrong/stale sequence is rejected.

func (*Service) Replay 

func (s *Service) Replay(ctx context.Context, id ID, sequence events.SequenceID) (jobs.Handle, error)

Replay explicitly requests replay and returns the actual server-issued job handle. Acceptance is neither job completion nor observer catch-up. Unlike C#, an invalid/empty JobId fails with OutcomeUnknownError rather than JobId.NotSet. Sequence must be explicit; use the observer's declared source sequence.

Example 
package main

import (
	"context"
	"errors"
	"fmt"

	jobcontracts "github.com/cratis/chronicle.go/contracts/jobs"
	contracts "github.com/cratis/chronicle.go/contracts/observation"
	"github.com/cratis/chronicle.go/events"
	"github.com/cratis/chronicle.go/jobs"
	"github.com/cratis/chronicle.go/observation"
	"google.golang.org/grpc"
)

// exampleKernel is an offline fixture. In an application use store.Observers()
// and store.Jobs(), which enforce connection and artifact registration barriers.
type exampleKernel struct{}

func (exampleKernel) Invoke(_ context.Context, method string, _ any, reply any, _ ...grpc.CallOption) error {
	switch method {
	case contracts.Observers_Replay_FullMethodName:
		reply.(*contracts.ReplayResponse).JobId = "00112233-4455-6677-8899-aabbccddeeff"
	case jobcontracts.Jobs_AllJobs_FullMethodName:
		reply.(*jobcontracts.QueryResult_IEnumerable_JobSummaryResponse).IsAuthorized = true
	case contracts.Observers_WaitForCompletion_FullMethodName:
		reply.(*contracts.WaitForObserverCompletionResponse).IsSuccess = true
	default:
		return fmt.Errorf("unexpected example RPC %s", method)
	}
	return nil
}
func (exampleKernel) NewStream(context.Context, *grpc.StreamDesc, string, ...grpc.CallOption) (grpc.ClientStream, error) {
	return nil, fmt.Errorf("no example streams")
}

func main() {
	observers, err := observation.New("orders", "Default", exampleKernel{})
	if err != nil {
		panic(err)
	}
	ctx := context.Background()
	job, err := observers.Replay(ctx, "orders-summary", events.EventLog)
	if err != nil {
		panic(err)
	} // A lost reply may mean the replay was accepted. Do not blindly retry.
	fmt.Println("accepted job:", job.ID())
	_, err = job.WaitForCompletion(ctx, 0)
	var absent *jobs.AbsentError
	if errors.As(err, &absent) {
		fmt.Println("job absent: completion unknown")
	} else if err != nil {
		panic(err)
	}
}

Output:
accepted job: 00112233-4455-6677-8899-aabbccddeeff
job absent: completion unknown

func (*Service) ReplayPartition 

func (s *Service) ReplayPartition(ctx context.Context, id ID, sequence events.SequenceID, partition Partition) error

ReplayPartition explicitly requests one partition replay. The wire returns no job identity; nil means acknowledgement only, not replay completion.

func (*Service) RetryPartition 

func (s *Service) RetryPartition(ctx context.Context, id ID, sequence events.SequenceID, partition Partition) (RecoveryOutcome, error)

RetryPartition explicitly requests recovery of one failed partition without clearing either quarantine. No automatic retry follows refusal or lost replies.

func (*Service) WaitForCompletion 

func (s *Service) WaitForCompletion(ctx context.Context, c Completion, timeout time.Duration) (CompletionResult, error)

WaitForCompletion waits for observers of the original appended event types. The kernel also reports success when there are no matching observers. This is not producer readiness or durable sink persistence evidence. The service's store/namespace must match; sequence comes ONLY from completion. Nil service with a committed tail returns CannotWaitError. Zero timeout selects 5s server + 200ms client grace. InfiniteTimeout sends 0 and adds no deadline. Caller cancellation/deadline remains an error; server/client-budget expiration is a TimedOut result. No RPC/mutation is retried and no goroutine is started.

Example 
package main

import (
	"context"
	"fmt"

	jobcontracts "github.com/cratis/chronicle.go/contracts/jobs"
	contracts "github.com/cratis/chronicle.go/contracts/observation"
	"github.com/cratis/chronicle.go/events"
	"github.com/cratis/chronicle.go/observation"
	"google.golang.org/grpc"
)

// exampleKernel is an offline fixture. In an application use store.Observers()
// and store.Jobs(), which enforce connection and artifact registration barriers.
type exampleKernel struct{}

func (exampleKernel) Invoke(_ context.Context, method string, _ any, reply any, _ ...grpc.CallOption) error {
	switch method {
	case contracts.Observers_Replay_FullMethodName:
		reply.(*contracts.ReplayResponse).JobId = "00112233-4455-6677-8899-aabbccddeeff"
	case jobcontracts.Jobs_AllJobs_FullMethodName:
		reply.(*jobcontracts.QueryResult_IEnumerable_JobSummaryResponse).IsAuthorized = true
	case contracts.Observers_WaitForCompletion_FullMethodName:
		reply.(*contracts.WaitForObserverCompletionResponse).IsSuccess = true
	default:
		return fmt.Errorf("unexpected example RPC %s", method)
	}
	return nil
}
func (exampleKernel) NewStream(context.Context, *grpc.StreamDesc, string, ...grpc.CallOption) (grpc.ClientStream, error) {
	return nil, fmt.Errorf("no example streams")
}

func main() {
	observers, err := observation.New("orders", "Default", exampleKernel{})
	if err != nil {
		panic(err)
	}
	// Normally obtain this immutable value from AppendManyWithMetadata(...).Completion().
	// These are the ORIGINAL append coordinates and each original input's position.
	completion, err := observation.NewCompletion("orders", "Default", events.EventLog,
		[]events.TypeRef{{ID: "OrderPlaced", Generation: 1}, {ID: "OrderPaid", Generation: 1}, {ID: "OrderPlaced", Generation: 1}},
		[]events.SequenceNumber{3, 5, 8})
	if err != nil {
		panic(err)
	}
	result, err := observers.WaitForCompletion(context.Background(), completion, 0)
	if err != nil {
		panic(err)
	}
	fmt.Println("processed:", result.IsSuccess(), "timed out:", result.TimedOut())
	fmt.Println("OrderPaid tail:", completion.Target().EventTypeTails["OrderPaid"])
}

Output:
processed: true timed out: false
OrderPaid tail: 5

type Type 

type Type int32

Type identifies the observer's kind, preserving unknown numeric values. 

const (
	// UnknownType means no kind is reported.
	UnknownType Type = iota
	// Reactor is a client-side event handler.
	Reactor
	// Projection is a kernel projection.
	Projection
	// Reducer is a client-side fold.
	Reducer
	// External is an external observer.
	External
)

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.