Documentation
¶
Overview ¶
Package dtmx defines Kernel's distributed transaction abstraction.
dtmx is intentionally provider-neutral. Business packages depend on this package only, while concrete implementations live in subpackages such as dtmx/dtm. The first production implementation is backed by dtm-labs/dtm via the official Go HTTP SDK.
Import an implementation once during bootstrap:
import _ "github.com/aisphereio/kernel/dtmx/dtm"
Then open it from config:
manager, err := dtmx.New(dtmx.Config{
Enabled: true,
Driver: "dtm",
Server: "http://127.0.0.1:36789/api/dtmsvr",
})
Kernel services should model cross-resource workflows as Saga branches with idempotent action and compensation endpoints. Large binary payloads must not be put into DTM request bodies; put objects into object storage first and pass only object keys, hashes, sizes, and metadata through Saga payloads.
Index ¶
- Constants
- Variables
- func BranchAuthMiddleware(secret string, next http.Handler) http.Handler
- func BuildBranchURL(base, prefix, branchPath string) string
- func Inject(ctx context.Context, manager Manager) context.Context
- func IsDriverRegistered(name string) bool
- func RegisterDriver(name string, opener DriverOpener)
- func RegisteredDrivers() []string
- func ValidateBranchRequest(r *http.Request, secret string) error
- func ValidateBranchToken(expected, got string) error
- type Config
- type DriverOpener
- type Manager
- type Saga
- type SagaOption
- type SagaOptions
- type SagaStep
- type TransactionResult
Constants ¶
const ( DriverDTM = "dtm" ProtocolHTTP = "http" )
const ( CodeDisabled = errorx.Code("DTMX_DISABLED") CodeInvalidConfig = errorx.Code("DTMX_INVALID_CONFIG") CodeUnknownDriver = errorx.Code("DTMX_UNKNOWN_DRIVER") CodeInvalidSaga = errorx.Code("DTMX_INVALID_SAGA") CodeUnsupportedProtocol = errorx.Code("DTMX_UNSUPPORTED_PROTOCOL") CodeSubmitFailed = errorx.Code("DTMX_SUBMIT_FAILED") CodeNewGIDFailed = errorx.Code("DTMX_NEW_GID_FAILED") CodeTimeout = errorx.Code("DTMX_TIMEOUT") CodeCanceled = errorx.Code("DTMX_CANCELED") )
const ( // BranchAuthHeader is the default header used to authenticate DTM -> service // branch callbacks. dtmx/dtm sends this header when Config.BranchSecret is // configured, and BranchAuthMiddleware validates it on callback handlers. BranchAuthHeader = "X-Kernel-DTM-Branch-Token" )
Variables ¶
var ( ErrDisabled = errors.New("dtmx: distributed transaction is disabled") ErrInvalidConfig = errors.New("dtmx: invalid config") ErrUnknownDriver = errors.New("dtmx: unknown driver") ErrNoSteps = errors.New("dtmx: saga has no steps") ErrInvalidBranch = errors.New("dtmx: invalid saga branch") ErrGIDRequired = errors.New("dtmx: gid is required") ErrUnsupportedProtocol = errors.New("dtmx: unsupported transaction protocol") ErrSubmitFailed = errors.New("dtmx: submit transaction failed") ErrNewGIDFailed = errors.New("dtmx: create gid failed") )
Functions ¶
func BranchAuthMiddleware ¶ added in v0.0.9
BranchAuthMiddleware wraps internal DTM branch endpoints with shared-secret validation. It writes 401 and does not call next when validation fails.
func BuildBranchURL ¶ added in v0.0.9
func IsDriverRegistered ¶ added in v0.0.9
func RegisterDriver ¶ added in v0.0.9
func RegisterDriver(name string, opener DriverOpener)
func RegisteredDrivers ¶ added in v0.0.9
func RegisteredDrivers() []string
func ValidateBranchRequest ¶ added in v0.0.9
ValidateBranchRequest validates the default branch-auth header on an HTTP request. It is intentionally net/http only so kernel transport adapters and business services can use it without importing the DTM SDK.
func ValidateBranchToken ¶ added in v0.0.9
ValidateBranchToken validates a received DTM branch token using constant-time comparison. Empty expected means branch authentication is disabled; this is allowed for local development, but production services should normally set a BranchSecret or protect branch endpoints with mTLS/private networking.
Types ¶
type Config ¶
type Config struct {
// Enabled controls whether distributed transactions are active. When false,
// New returns a disabled no-op Manager that fails fast on SubmitSaga.
Enabled bool `json:"enabled" yaml:"enabled"`
// Driver selects the registered implementation. The production driver is
// "dtm" from package github.com/aisphereio/kernel/dtmx/dtm.
Driver string `json:"driver" yaml:"driver"`
// Protocol selects how DTM calls branch endpoints. Currently "http" is
// implemented; gRPC can be added without changing business code.
Protocol string `json:"protocol" yaml:"protocol"`
// Server is the DTM HTTP API endpoint, e.g.
// http://127.0.0.1:36789/api/dtmsvr.
Server string `json:"server" yaml:"server"`
// ServiceBaseURL is the externally reachable base URL for this service.
// Helpers use it to build branch callback URLs.
ServiceBaseURL string `json:"service_base_url" yaml:"service_base_url"`
// BranchPrefix is the prefix for internal branch endpoints, e.g.
// /internal/dtm. It is optional and used only by BranchURL helper.
BranchPrefix string `json:"branch_prefix" yaml:"branch_prefix"`
// WaitResult asks DTM to wait for Saga execution result on Submit. Use true
// in synchronous user-facing flows; false in fire-and-retry background flows.
WaitResult bool `json:"wait_result" yaml:"wait_result"`
// Timeout bounds the local SubmitSaga call. DTM itself still owns branch
// retry/timeout behavior.
Timeout time.Duration `json:"timeout_ns" yaml:"timeout_ns"`
// BranchSecret is propagated to DTM as a branch header. DTM then forwards it
// to branch action/compensate endpoints, where ValidateBranchRequest or
// BranchAuthMiddleware can verify the callback. Keep it empty only when the
// branch endpoints are protected by stronger network controls such as mTLS
// and private networking.
BranchSecret string `json:"branch_secret" yaml:"branch_secret"`
// Logger is the component logger. If nil, dtmx uses logx.DefaultLogger().
Logger logx.Logger `json:"-" yaml:"-"`
// Metrics is the optional metrics manager.
Metrics metricsx.Manager `json:"-" yaml:"-"`
// MetricsEnabled controls whether dtmx records metrics.
MetricsEnabled bool `json:"metrics_enabled" yaml:"metrics_enabled"`
}
Config configures Kernel distributed transactions.
This config is for the application-side transaction manager client. The DTM server itself is an external process managed by deployment; kernel only needs the DTM API endpoint and service callback address.
type DriverOpener ¶ added in v0.0.9
type Manager ¶ added in v0.0.9
type Manager interface {
Enabled() bool
DriverName() string
NewGID(ctx context.Context) (string, error)
SubmitSaga(ctx context.Context, saga Saga) (TransactionResult, error)
BranchURL(path string) string
Close() error
}
Manager is the business-facing distributed transaction interface.
Business modules should depend on Manager, Saga, and SagaStep only. They should not import github.com/dtm-labs/client directly.
func FromContext ¶ added in v0.0.9
func FromContextOr ¶ added in v0.0.9
type Saga ¶
type Saga struct {
GID string
Name string
Steps []SagaStep
// Options override Config defaults for a single Saga.
Options SagaOptions
}
func NewSaga ¶
func NewSaga(gid, name string, opts ...SagaOption) Saga
type SagaOption ¶ added in v0.0.9
type SagaOption func(*Saga)
func WithSagaBranchHeader ¶ added in v0.0.9
func WithSagaBranchHeader(key, value string) SagaOption
WithSagaBranchHeader adds a header that DTM forwards to every branch action/compensate request for this Saga. Prefer Config.BranchSecret for the common shared-secret case; use this for per-transaction metadata only.
func WithSagaBranchHeaders ¶ added in v0.0.9
func WithSagaBranchHeaders(headers map[string]string) SagaOption
WithSagaBranchHeaders adds headers that DTM forwards to every branch action/compensate request for this Saga.
func WithSagaTimeout ¶ added in v0.0.9
func WithSagaTimeout(timeout time.Duration) SagaOption
func WithSagaWaitResult ¶ added in v0.0.9
func WithSagaWaitResult(wait bool) SagaOption