Documentation
¶
Overview ¶
Package safety provides end-to-end data protection and deterministic queuing primitives for safety-oriented DDS deployments.
E2EPublisher prepends an 18-byte protection header to every payload before writing. E2ESubscriber strips the header and validates CRC, sequence counter, and sample freshness on every received sample.
Wire format (little-endian, 18 bytes followed by original payload):
Bytes 0–1 DataID (uint16) Bytes 2–3 SourceID (uint16) Bytes 4–7 SequenceCounter (uint32, monotonically increasing per publisher) Bytes 8–15 Timestamp (int64, Unix nanoseconds at time of Write) Bytes 16–17 CRC-16/CCITT-FALSE over bytes 0–15 plus the original payload Bytes 18+ Original payload
The CRC slot is treated as zero when computing the CRC.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrQueueFull = errors.New("safety: queue full")
ErrQueueFull is returned by Enqueue when the DeterministicQueue has no remaining capacity.
Functions ¶
This section is empty.
Types ¶
type DeterministicQueue ¶
type DeterministicQueue struct {
// contains filtered or unexported fields
}
DeterministicQueue is a bounded FIFO that decouples application writes from the underlying publisher. It contains the publisher's panic (converting it to an error on the Errors channel) and provides non-blocking back-pressure via ErrQueueFull.
Typical use:
q := safety.NewDeterministicQueue(pub, 128).Start()
defer q.Stop()
if err := q.Enqueue(payload); err != nil {
// handle back-pressure
}
func NewDeterministicQueue ¶
func NewDeterministicQueue(pub dds.Publisher, depth int) *DeterministicQueue
NewDeterministicQueue creates a queue with the given channel depth that drains into pub. Call Start to begin delivery. A depth of ≤ 0 uses 64.
func (*DeterministicQueue) Enqueue ¶
func (q *DeterministicQueue) Enqueue(payload []byte) error
Enqueue adds payload to the queue without blocking. Returns ErrQueueFull if the queue is at capacity.
func (*DeterministicQueue) Errors ¶
func (q *DeterministicQueue) Errors() <-chan error
Errors returns a channel that receives errors from the drain goroutine, including publisher errors and panics wrapped as errors.
func (*DeterministicQueue) Start ¶
func (q *DeterministicQueue) Start() *DeterministicQueue
Start launches the drain goroutine. Returns q for chaining.
func (*DeterministicQueue) Stop ¶
func (q *DeterministicQueue) Stop()
Stop signals the drain goroutine to exit and waits for it to finish. Safe to call more than once.
type E2EConfig ¶
type E2EConfig struct {
// DataID identifies the logical data element (0–65535).
DataID uint16
// SourceID identifies the sender (0–65535).
SourceID uint16
// MaxAge is the maximum permitted age of a received sample, measured from
// the timestamp written by the publisher. Zero disables freshness checking.
MaxAge time.Duration
// Topic is the DDS topic name; used to label safety metrics. Optional.
Topic string
}
E2EConfig configures end-to-end protection parameters shared by a publisher and subscriber pair.
type E2EError ¶
type E2EError struct {
Kind E2EErrorKind
Counter uint32
Message string
}
E2EError is sent over the Errors channel when a safety check fails.
type E2EErrorKind ¶
type E2EErrorKind int
E2EErrorKind categorises safety check failures reported by E2ESubscriber.
const ( // ErrCRCMismatch means the CRC in the frame header did not match. ErrCRCMismatch E2EErrorKind = iota // ErrSequenceGap means one or more sequence numbers were skipped. ErrSequenceGap // ErrStaleSample means the sample's timestamp is older than MaxAge. ErrStaleSample // ErrHeaderTooShort means the payload is shorter than the 18-byte header. ErrHeaderTooShort )
type E2EPublisher ¶
type E2EPublisher struct {
// contains filtered or unexported fields
}
E2EPublisher wraps a dds.Publisher and prepends an E2E protection header to every Write payload. It satisfies dds.Publisher.
func NewE2EPublisher ¶
func NewE2EPublisher(pub dds.Publisher, cfg E2EConfig) *E2EPublisher
NewE2EPublisher wraps pub with end-to-end protection configured by cfg.
func (*E2EPublisher) Close ¶
func (p *E2EPublisher) Close() error
Close closes the underlying publisher.
func (*E2EPublisher) Write ¶
func (p *E2EPublisher) Write(payload []byte) error
Write prepends an E2E header to payload and writes the frame to the underlying publisher.
type E2ESubscriber ¶
type E2ESubscriber struct {
// contains filtered or unexported fields
}
E2ESubscriber wraps a dds.Subscriber, strips the E2E header from each received sample, and validates CRC, sequence counter, and freshness. Valid samples are forwarded to C(); failures are reported on Errors().
The background pump goroutine starts on construction and exits when Close is called or the underlying subscriber channel is closed.
E2ESubscriber implements SafetyMetricsProvider; call SafetyMetrics() to retrieve cumulative violation counters, and SafetyEvents() to receive a channel of individual SafetyEvent values (e.g., for monitor integration).
func NewE2ESubscriber ¶
func NewE2ESubscriber(sub dds.Subscriber, cfg E2EConfig) *E2ESubscriber
NewE2ESubscriber wraps sub with E2E validation using cfg.
func (*E2ESubscriber) C ¶
func (s *E2ESubscriber) C() <-chan dds.Sample
C returns the channel that delivers validated, header-stripped samples. It satisfies dds.Subscriber.
func (*E2ESubscriber) Close ¶
func (s *E2ESubscriber) Close() error
Close stops the pump goroutine and closes the underlying subscriber.
func (*E2ESubscriber) Errors ¶
func (s *E2ESubscriber) Errors() <-chan E2EError
Errors returns a channel that receives E2EError values for every safety check failure. The channel is buffered (32); overflows are silently dropped.
func (*E2ESubscriber) SafetyEvents ¶ added in v0.12.1
func (s *E2ESubscriber) SafetyEvents() <-chan SafetyEvent
SafetyEvents returns a channel that receives one SafetyEvent per violation. The channel is buffered (64); slow consumers cause events to be dropped. Intended for integration with monitor.Monitor.WatchSafety.
func (*E2ESubscriber) SafetyMetrics ¶ added in v0.12.1
func (s *E2ESubscriber) SafetyMetrics() Snapshot
SafetyMetrics returns a point-in-time snapshot of cumulative violation counters. It implements SafetyMetricsProvider.
type RateAlert ¶ added in v0.13.0
type RateAlert struct {
// Topic is the DDS topic for which the threshold was exceeded.
Topic string
// Kind is the safety event category that triggered the alert.
Kind SafetyEventKind
// Rate is the observed violation rate (events per second) over the last window.
Rate float64
// Threshold is the configured maximum rate that was exceeded.
Threshold float64
}
RateAlert is emitted by RateMonitor when a violation rate exceeds its threshold.
type RateMonitor ¶ added in v0.13.0
type RateMonitor struct {
// contains filtered or unexported fields
}
RateMonitor polls a SafetyMetricsProvider at a fixed interval and invokes an alert callback whenever any violation rate exceeds its configured threshold. It is safe for concurrent use.
func NewRateMonitor ¶ added in v0.13.0
func NewRateMonitor(p SafetyMetricsProvider, interval time.Duration, thresholds RateThresholds, onAlert func(RateAlert)) *RateMonitor
NewRateMonitor starts a RateMonitor that polls p every interval. When a violation rate exceeds a threshold, onAlert is called synchronously from the background goroutine. Call Stop to release resources. Intervals ≤ 0 default to 5 seconds.
func (*RateMonitor) Stop ¶ added in v0.13.0
func (rm *RateMonitor) Stop()
Stop shuts down the background polling goroutine. It is safe to call once.
type RateThresholds ¶ added in v0.13.0
type RateThresholds struct {
CRCFailureRate float64
SequenceGapRate float64
StaleSampleRate float64
SchemaViolationRate float64
}
RateThresholds defines the maximum tolerated violation rate (events per second) for each safety event kind. A threshold of zero disables that check.
type SafetyEvent ¶ added in v0.12.1
type SafetyEvent struct {
// Kind classifies the violation.
Kind SafetyEventKind
// Topic is the DDS topic on which the violation was detected.
Topic string
// Counter is the sequence counter from the frame header (0 if unavailable).
Counter uint32
// Message is a human-readable description of the violation.
Message string
}
SafetyEvent represents a single E2E safety violation detected by an E2ESubscriber. It is broadcast to registered safety event listeners (e.g., monitor.Monitor).
type SafetyEventKind ¶ added in v0.12.1
type SafetyEventKind int
SafetyEventKind categorises safety events emitted by E2ESubscriber.
const ( // SafetyEventCRCFailure is emitted when a frame fails its CRC check. SafetyEventCRCFailure SafetyEventKind = iota // SafetyEventSequenceGap is emitted when one or more sequence numbers are skipped. SafetyEventSequenceGap // SafetyEventStaleSample is emitted when a sample exceeds MaxAge. SafetyEventStaleSample // SafetyEventHeaderTooShort is emitted when a payload is shorter than the E2E header. SafetyEventHeaderTooShort // SafetyEventSchemaViolation is emitted when a sample's type does not match the registered schema. SafetyEventSchemaViolation )
func (SafetyEventKind) String ¶ added in v0.12.1
func (k SafetyEventKind) String() string
type SafetyMetricsProvider ¶ added in v0.12.1
type SafetyMetricsProvider interface {
// SafetyMetrics returns a point-in-time snapshot of safety violation counters.
SafetyMetrics() Snapshot
}
SafetyMetricsProvider is implemented by components that expose E2E violation counters.
type Snapshot ¶ added in v0.12.1
type Snapshot struct {
Topic string `json:"topic"`
CRCFailures uint64 `json:"crc_failures"`
SequenceGaps uint64 `json:"sequence_gaps"`
StaleSamples uint64 `json:"stale_samples"`
HeaderTooShort uint64 `json:"header_too_short"`
SchemaViolations uint64 `json:"schema_violations"`
ValidSamples uint64 `json:"valid_samples"`
}
Snapshot is a point-in-time copy of violation counters for one subscriber.
Source Files