Documentation
¶
Overview ¶
Package xcom implements the Dexaflow XCom subsystem: small typed payloads passed between tasks, stored in Redis with a hard size limit and a TTL, with metadata indexed in Postgres for retrieval (ADR 0006).
Index ¶
- Constants
- Variables
- type Backend
- type BatchBackend
- type Entry
- type Index
- type IndexEntry
- type Key
- type PostgresBackend
- func (b *PostgresBackend) Delete(ctx context.Context, key string) error
- func (b *PostgresBackend) DeleteExpired(ctx context.Context) (int64, error)
- func (b *PostgresBackend) Fetch(ctx context.Context, key string) (Entry, error)
- func (b *PostgresBackend) FetchMany(ctx context.Context, keys []string) ([]Entry, []bool, error)
- func (b *PostgresBackend) List(ctx context.Context, pattern string) ([]string, error)
- func (b *PostgresBackend) Push(ctx context.Context, key string, entry Entry, ttl time.Duration) error
- type RedisBackend
- func (b *RedisBackend) Delete(ctx context.Context, key string) error
- func (b *RedisBackend) Fetch(ctx context.Context, key string) (Entry, error)
- func (b *RedisBackend) FetchMany(ctx context.Context, keys []string) ([]Entry, []bool, error)
- func (b *RedisBackend) List(ctx context.Context, pattern string) ([]string, error)
- func (b *RedisBackend) Push(ctx context.Context, key string, entry Entry, ttl time.Duration) error
- type Result
- type Service
Constants ¶
const DefaultTTL = 7 * 24 * time.Hour
DefaultTTL is the default XCom lifetime, configurable per DAG.
const MaxSizeBytes = 256 * 1024
MaxSizeBytes is the hard per-key XCom size limit (256 KB). Larger payloads must use a blob store and pass a reference (ADR 0006).
Variables ¶
var ( // ErrNotFound is returned when a key is absent or expired. ErrNotFound = errors.New("xcom not found") // ErrTooLarge is returned when a payload exceeds MaxSizeBytes. ErrTooLarge = errors.New("xcom payload exceeds the 256KB limit") // ErrSchemaMismatch is returned when a payload violates its declared schema. ErrSchemaMismatch = errors.New("xcom payload does not match the declared schema") )
Sentinel errors returned by the XCom subsystem.
Functions ¶
This section is empty.
Types ¶
type Backend ¶
type Backend interface {
Push(ctx context.Context, key string, entry Entry, ttl time.Duration) error
Fetch(ctx context.Context, key string) (Entry, error)
Delete(ctx context.Context, key string) error
List(ctx context.Context, pattern string) ([]string, error)
}
Backend stores XCom entries with a TTL. RedisBackend is the production implementation; tests use a fake.
type BatchBackend ¶ added in v0.5.1
type BatchBackend interface {
FetchMany(ctx context.Context, keys []string) (entries []Entry, found []bool, err error)
}
BatchBackend is a Backend that reads several keys in one round trip. found[i] is false when keys[i] is absent or expired.
type Entry ¶
type Entry struct {
Value []byte `json:"value"`
ContentType string `json:"content_type"`
SizeBytes int `json:"size_bytes"`
CreatedAt time.Time `json:"created_at"`
}
Entry is a stored XCom value together with its metadata.
type Index ¶
type Index interface {
RecordXCom(ctx context.Context, entry IndexEntry) error
}
Index records XCom metadata for retrieval and cleanup (Postgres-backed).
type IndexEntry ¶
type IndexEntry struct {
TenantID string
RunID string
TaskID string
Name string
RedisKey string
SizeBytes int
ContentType string
ExpiresAt time.Time
}
IndexEntry is the Postgres-side metadata recorded for each push, used for UI listing and expiry-based cleanup.
type Key ¶
Key identifies an XCom value. Its string form is the Redis key (ADR 0006): xcom:{tenant_id}:{dag_id}:{run_id}:{task_id}:{key_name}.
type PostgresBackend ¶
type PostgresBackend struct {
// contains filtered or unexported fields
}
PostgresBackend is the XCom Backend used by Dexaflow Lite: it stores entries in the xcom_store table with an expires_at TTL, so Lite needs no Redis at all (scheduler locks already use Postgres advisory locks). Production keeps the RedisBackend per ADR 0006.
func NewPostgresBackend ¶
func NewPostgresBackend(db pgxQuerier) *PostgresBackend
NewPostgresBackend builds a PostgresBackend over the given pgx pool.
func (*PostgresBackend) Delete ¶
func (b *PostgresBackend) Delete(ctx context.Context, key string) error
Delete removes the entry at key.
func (*PostgresBackend) DeleteExpired ¶
func (b *PostgresBackend) DeleteExpired(ctx context.Context) (int64, error)
DeleteExpired removes rows past their TTL, returning how many were deleted. It is the periodic sweep that replaces Redis's native key expiry.
func (*PostgresBackend) Fetch ¶
Fetch returns the entry at key, or ErrNotFound when it is absent or expired.
func (*PostgresBackend) FetchMany ¶ added in v0.5.1
FetchMany reads every key with one query, in order. Absent or expired keys come back as not found; a key requested twice is answered twice.
type RedisBackend ¶
type RedisBackend struct {
// contains filtered or unexported fields
}
RedisBackend is the production XCom Backend: it stores each entry as a JSON envelope under its key with a TTL, so Redis expires values natively (ADR 0006).
func NewRedisBackend ¶
func NewRedisBackend(client *redis.Client) *RedisBackend
NewRedisBackend builds a RedisBackend over the given go-redis client.
func (*RedisBackend) Delete ¶
func (b *RedisBackend) Delete(ctx context.Context, key string) error
Delete removes the entry at key.
func (*RedisBackend) Fetch ¶
Fetch returns the entry at key, or ErrNotFound when it is absent or expired.
func (*RedisBackend) FetchMany ¶ added in v0.5.1
FetchMany reads every key with one MGET, in order. Absent or expired keys come back as not found.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service validates and stores XCom values, recording their metadata in the index.
func NewService ¶
NewService builds an XCom service over the given backend and index, applying the given TTL to stored values.
Source Files