Documentation
¶
Overview ¶
Package pipeline provides the TDTP v1.4 consumer pre-flight verification pipeline: Mercury hash registry check → local xxh3 integrity → Dictionary expansion, with configurable fallback when Mercury is unavailable.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ComputeAndRegisterIntegrity ¶
func ComputeAndRegisterIntegrity(ctx context.Context, pkt *packet.DataPacket, client HashRegistrar, sender string) error
ComputeAndRegisterIntegrity stamps pkt with TDTP v1.4 xxh3_128 integrity hashes (packet.ComputeIntegrity) and, when client is non-nil, registers the packet fingerprint with xZMercury's hash registry so a consumer's VerifyAndPrepare can later confirm nothing was tampered with in transit.
Producer-side counterpart to VerifyAndPrepare in this same package.
Mandatory for TDTP v1.5 encryption, not just an opt-in v1.4 feature: once a packet's Version is >= "1.4" (checked via packet.NeedsRowCountCheck), VerifyAndPrepare's Mercury pre-flight ALWAYS runs on the consumer side — runMercuryCheck treats an empty pkt.XXH3 as ErrHashNotRegistered (a hard block), not "integrity wasn't requested, skip". A v1.5-encrypted packet that skipped this call would therefore be unimportable the moment --mercury-url is set — which v1.5 decryption itself always requires. See docs/tdtp-protocol-schema.md → "v1.5" for the full design; this function exists specifically to close that gap, not as a general-purpose convenience — every TDTP v1.5 encryption call site must call this before compression, or produce packets that cannot be imported.
Order matters and is fixed, same as compression/encryption: this must run BEFORE compression (hashes cover plaintext row values) and BEFORE encryption (packet.EncryptSections later bumps Version to "1.5", overwriting the "1.4" this function sets — the final packet ends up correctly versioned either way).
Types ¶
type FallbackPolicy ¶
type FallbackPolicy int
FallbackPolicy controls consumer behaviour when xzMercury is unreachable.
const ( // FallbackBlock stops processing — no packet accepted without Mercury // confirmation. Most secure; use for financial, medical, audit pipelines. // Mercury down → ErrMercuryUnavailable returned to caller. FallbackBlock FallbackPolicy = iota // FallbackDegrade skips the Mercury check and relies on local xxh3 hashes // only. No executor control — any producer can forge a packet — but data // integrity (schema + rows unchanged) is still verified locally. // Use when availability > security (operational continuity required). FallbackDegrade // FallbackDowngrade converts the packet to v1.3.1 in-place (expands // Dictionary tokens, clears integrity fields, sets Version="1.3.1") and // processes it via the legacy path. Zero Mercury dependency after downgrade. // Use when the downstream system only supports v1.3.1 or Mercury SLA is low. FallbackDowngrade )
type HashRegistrar ¶
type HashRegistrar interface {
RegisterHash(ctx context.Context, uuid string, part int, xxh3, tableName, sender, packetVersion string) error
}
HashRegistrar is satisfied by *mercury.Client (production) and by any dev/test substitute — mirrors the same optional-override shape processors.MercuryBinder already provides for BindKey, so callers that support a MercuryBinder override for key operations (e.g. pkg/etl's Exporter) can support one here too without a parallel mechanism.
type HashVerifier ¶
type HashVerifier interface {
VerifyHash(ctx context.Context, uuid string, part int, xxh3, version string) (*mercury.HashRecord, error)
}
HashVerifier abstracts xzMercury so the pipeline can be tested without a live Mercury instance. mercury.Client satisfies this interface.
type VerifyResult ¶
type VerifyResult struct {
// Version is the effective protocol version after optional downgrade.
// "1.4" = full verification passed; "1.3.1" = downgraded due to fallback.
Version string
// Degraded is true when Mercury was unavailable and a non-blocking fallback
// was applied. Always log this; alert if it persists.
Degraded bool
// DegradedReason explains why degraded mode was triggered.
DegradedReason string
// MercuryRecord is the hash record returned by Mercury on successful
// verification. Nil when Degraded=true or packet is pre-1.4.
MercuryRecord *mercury.HashRecord
// VerifiedAt is the timestamp of successful verification.
VerifiedAt time.Time
}
VerifyResult is returned by VerifyAndPrepare.
func VerifyAndPrepare ¶
func VerifyAndPrepare( ctx context.Context, pkt *packet.DataPacket, verifier HashVerifier, policy FallbackPolicy, ) (*VerifyResult, error)
VerifyAndPrepare is the single entry point for TDTP consumer pre-flight.
For v1.4 packets it runs:
Mercury executor check — UUID+part → stored_xxh3 == pkt.XXH3? Attacker cannot forge: Mercury slot is SET NX by authenticated producer.
Local xxh3 integrity — recompute schema/data/packet hashes locally. Detects in-transit corruption even without Mercury.
Dictionary expansion — expand @tokens in rows (transparent to caller).
When Mercury is unreachable the policy controls behaviour:
- FallbackBlock → return ErrMercuryUnavailable (fail-closed)
- FallbackDegrade → skip step 1, run steps 2+3 (degraded mode)
- FallbackDowngrade → Downgrade pkt to v1.3.1, skip steps 1+2+3
For pre-v1.4 packets all steps are skipped (legacy pass-through).
Source Files