quality

package
v0.0.1-beta.7 Latest Latest
Warning

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

 Go to latest Published: Sep 28, 2026 License: AGPL-3.0 Imports: 20 Imported by: 0

Documentation

Overview

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

Index

Constants

View Source 
const (
	DefaultRetiredCap        = 65536
	DefaultPendingCapBytes   = 256 * 1024 * 1024
	DefaultMinuteBucketsCap  = 4096
	EstimatedCellBytes       = 256
	EstimatedQualityRowBytes = 512
	EstimatedFlowMinuteBytes = 4096
	EstimatedFlowRowBytes    = 512
	DefaultFlowRowCap        = DefaultPendingCapBytes / EstimatedFlowRowBytes
)
View Source 
const (
	ErrClassNone    = -1
	ErrClass429     = 0
	ErrClass5xx     = 1
	ErrClassNetwork = 2
	ErrClass4xx     = 3
)

Variables

View Source 
var ErrCapacity = errors.New("quality recorder capacity exceeded")
View Source 
var TTFTEdges = [10]int64{10, 25, 50, 100, 250, 500, 1000, 2000, 5000, 10000}

Functions

func FlowChainCapacityOverflow 

func FlowChainCapacityOverflow() int64

func FlowChainEnqueueOverflow 

func FlowChainEnqueueOverflow() int64

func FlowChainIncompleteObserved 

func FlowChainIncompleteObserved() int64

func GenerateInstanceSrc 

func GenerateInstanceSrc() string

func ProcessCrashLossUnobservable 

func ProcessCrashLossUnobservable() bool

func ResetFlowChainCountersForTest 

func ResetFlowChainCountersForTest()

func RoutingLateSnapshotDropped 

func RoutingLateSnapshotDropped() int64

RoutingLateSnapshotDropped 返回本进程因超出观测保留期而被拒并丢弃的快照数。 只增不减；调用方按前后差值判定（进程级计数器，不做测试重置）。

Types

type AttemptContext 

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

func (*AttemptContext) Cancel 

func (a *AttemptContext) Cancel()

func (*AttemptContext) Complete 

func (a *AttemptContext) Complete(success bool, ttftMs *int64, tokens, calls, images int64)

func (*AttemptContext) CompleteObservation 

func (a *AttemptContext) CompleteObservation(obs Observation)

func (*AttemptContext) IsCompleted 

func (a *AttemptContext) IsCompleted() bool

func (*AttemptContext) IsZero 

func (a *AttemptContext) IsZero() bool

type Cell 

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

type FlowMinute 

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

func (*FlowMinute) Clone 

func (f *FlowMinute) Clone() *FlowMinute

func (*FlowMinute) FlowRows 

func (f *FlowMinute) FlowRows() []repository.RoutingFlowRow

FlowRows 返回该分钟的行切片（内部存储，不做拷贝）。所有权契约：

  • 保留态分钟（Recorder/FlowOwner 持有）返回的行**只读**——调用方不得改写、 不得长期别名；需要独立副本时自行 clone；
  • 独占产物（snapshotForPG / redisPayload 的 materializeShell 输出，与该 分钟所有者不共享）可原地改写——flowRowsFromMinute 会写 InstanceSrc/AbsoluteSequence/TerminalMinute flush 戳——但必须同步消费后 立即丢弃该 *FlowMinute。

旧版逐次深拷贝访问器已删：发布/flush 路径全部走独占产物，逐次拷贝纯属多余 （压测 heap：FlowRows 深拷 0.69GB/战役），且拷贝掩盖了所有权语义。

func (*FlowMinute) HasFlowRows 

func (f *FlowMinute) HasFlowRows() bool

func (*FlowMinute) IsEmptySnapshot 

func (f *FlowMinute) IsEmptySnapshot() bool

func (*FlowMinute) Minute 

func (f *FlowMinute) Minute() int64

func (*FlowMinute) SetFlowRows 

func (f *FlowMinute) SetFlowRows(rows []repository.RoutingFlowRow)

type FlowOwner 

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

FlowOwner is the managed worker (worker.Worker + handler.StatsProvider) exclusively owning the folded flow state: the counter-cell table plus the per-minute cumulative shells. Constructed by NewRecorder (one per recorder); retrieved with Recorder.FlowOwner() for registration.

Lock discipline: o.mu guards the shells only (tick-owned reads and folds). The request path (FoldChain/Adds) never takes o.mu — only lock-free cell atomics plus the leaf shard locks inside the table.

func (*FlowOwner) Close 

func (o *FlowOwner) Close(ctx context.Context) error

Close stops the owner. Seal runs via SyncWorker.Close (unchanged); here only the lifecycle flips. Safe before Start and idempotent.

func (*FlowOwner) FoldChain 

func (o *FlowOwner) FoldChain(bucket int64, n int, next func(i int) (
	route domain.RouteClassIDVal,
	accountID, prevAccount, generation int64,
	ordinal uint8,
	lane, outcome, prevOutcome string,
	terminal, hasPrev bool,
))

FoldChain walks one completed chain's stack facts (option (a): terminality is final — the caller force-marks the last edge before walking) and folds each fact with one atomic Add. minuteBucket is minted ONCE by the caller and stamped on every fact, so all cells of the chain expand under one TerminalMinute. Cap-8: only the first 8 facts emit; the rest count cap-overflow with zero cell adds. Calls after owner close (or recorder finalization) land nowhere and count dropped.

Seal split (normative §4): per fact, the atomic Add carries the sealed-observed classification stamped at Add time (Go atomics are sequentially consistent, so the sealed-generation read carries at least acquire ordering). Pre-seal/post-seal races reconcile exactly once at fold time (drainFoldLocked pre-seal-origin-under-seal branch + seal sweep) — no request-side fixup exists by design (see the sealedGen field comment).

func (*FlowOwner) Name 

func (o *FlowOwner) Name() string

Name implements worker.Worker and handler.StatsProvider.

func (*FlowOwner) NoteCapOverflow 

func (o *FlowOwner) NoteCapOverflow()

NoteCapOverflow records one dropped attempt beyond the cap-8 bound at the append site: same unexported cap-overflow counter +1, zero cell adds.

func (*FlowOwner) NoteIncompleteChain 

func (o *FlowOwner) NoteIncompleteChain()

NoteIncompleteChain records a close-without-terminal (the §5.1 Close row: same unexported counter +1, facts emitted none).

func (*FlowOwner) SnapshotStats 

func (o *FlowOwner) SnapshotStats() FlowOwnerStats

func (*FlowOwner) Start 

func (o *FlowOwner) Start(ctx context.Context) error

Start marks the owner running. There is no owner goroutine anymore — nothing is queued, so no consumer loop exists; the tick expansions run inside the sync flushes. Registration order (owner before quality-sync) is unchanged.

func (*FlowOwner) Stats 

func (o *FlowOwner) Stats() any

type FlowOwnerStats 

type FlowOwnerStats struct {
	Running             bool  `json:"running"`
	Queued              int   `json:"queued"`
	QueueCap            int   `json:"queue_cap"`
	Accepted            int64 `json:"accepted"`
	Processed           int64 `json:"processed"`
	Overflowed          int64 `json:"overflowed"`
	EdgeRowsAccepted    int64 `json:"edge_rows_accepted"`
	EdgeRowsDropped     int64 `json:"edge_rows_dropped"`
	PendingMinutes      int   `json:"pending_minutes"`
	PendingBytes        int64 `json:"pending_bytes"`
	LastMergeDurationMs int64 `json:"last_merge_duration_ms"`
	ResidualSubmissions int64 `json:"residual_submissions"`
	ResidualRows        int64 `json:"residual_rows"`
	CloseUnixMs         int64 `json:"close_unix_ms"`
}

FlowOwnerStats is the typed Stats() payload served on /api/admin/ops/workers under name "quality-flow-owner". Field set is pinned by openapi.yaml WorkerStatus.stats — rename = contract break. v3 re-sourcing (values, not shape): queued = accepted-not-yet-folded facts (no queue remains); queue_cap = total counter-cell capacity; accepted = landed facts (event units — one FoldChain call contributes its emitted fact count, so accepted == processed + residual_submissions at drained points); processed = accepted-origin facts folded by the tick.

type Key 

type Key struct {
	IdentityVersion int16
	RouteClassID    [32]byte
	QualityClassID  [32]byte
	Fingerprint     [32]byte
}

Key is the canonical routing identity of a quality cell/row: the routing identity-hash version plus the unique index (route_class_id, quality_class_id, candidate_fingerprint) minus the minute bucket. IdentityVersion stays part of the key although the DB column was dropped: it is the Redis wire and fold version guard, and the first byte of every identity hash.

func CanonicalKey 

func CanonicalKey(routeClassID, qualityClassID, fingerprint [32]byte) Key

type Observation 

type Observation struct {
	Success             bool
	TTFTMs              *int64
	InputTokens         int64
	OutputTokens        int64
	CacheReadTokens     int64
	CacheCreationTokens int64
	Tokens              int64
	Calls               int64
	Images              int64
	ErrClass            int
	IsCancel            bool
	IsLocal             bool
	IsReservation       bool
}

type PGQualityWriter 

type PGQualityWriter interface {
	UpsertQualityRow(ctx context.Context, row repository.RoutingQualityRow) error
	UpsertFlowSnapshot(ctx context.Context, instanceSrc string, terminalMinute time.Time, identityVersion int16, absoluteSequence int64, rows []repository.RoutingFlowRow) error
}

type QualityMinute 

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

func NewQualityMinute 

func NewQualityMinute(minute int64, key Key) *QualityMinute

func (*QualityMinute) Attempts 

func (q *QualityMinute) Attempts() int64

func (*QualityMinute) CacheCreateTokens 

func (q *QualityMinute) CacheCreateTokens() int64

func (*QualityMinute) CacheReadTokens 

func (q *QualityMinute) CacheReadTokens() int64

func (*QualityMinute) Calls 

func (q *QualityMinute) Calls() int64

func (*QualityMinute) Clone 

func (q *QualityMinute) Clone() *QualityMinute

func (*QualityMinute) Err4xx 

func (q *QualityMinute) Err4xx() int64

func (*QualityMinute) Err5xx 

func (q *QualityMinute) Err5xx() int64

func (*QualityMinute) Err429 

func (q *QualityMinute) Err429() int64

func (*QualityMinute) ErrNetwork 

func (q *QualityMinute) ErrNetwork() int64

func (*QualityMinute) Fingerprint 

func (q *QualityMinute) Fingerprint() [32]byte

func (*QualityMinute) Gen 

func (q *QualityMinute) Gen() uint64

func (*QualityMinute) Hist 

func (q *QualityMinute) Hist() [10]int64

func (*QualityMinute) IdentityVersion 

func (q *QualityMinute) IdentityVersion() int16

func (*QualityMinute) Images 

func (q *QualityMinute) Images() int64

func (*QualityMinute) InputTokens 

func (q *QualityMinute) InputTokens() int64

func (*QualityMinute) Key 

func (q *QualityMinute) Key() Key

func (*QualityMinute) Minute 

func (q *QualityMinute) Minute() int64

func (*QualityMinute) OutputTokens 

func (q *QualityMinute) OutputTokens() int64

func (*QualityMinute) QualityClassID 

func (q *QualityMinute) QualityClassID() [32]byte

func (*QualityMinute) RouteClassID 

func (q *QualityMinute) RouteClassID() [32]byte

func (*QualityMinute) SetAttempts 

func (q *QualityMinute) SetAttempts(v int64)

func (*QualityMinute) SetCacheCreateTokens 

func (q *QualityMinute) SetCacheCreateTokens(v int64)

func (*QualityMinute) SetCacheReadTokens 

func (q *QualityMinute) SetCacheReadTokens(v int64)

func (*QualityMinute) SetCalls 

func (q *QualityMinute) SetCalls(v int64)

func (*QualityMinute) SetErr4xx 

func (q *QualityMinute) SetErr4xx(v int64)

func (*QualityMinute) SetErr5xx 

func (q *QualityMinute) SetErr5xx(v int64)

func (*QualityMinute) SetErr429 

func (q *QualityMinute) SetErr429(v int64)

func (*QualityMinute) SetErrNetwork 

func (q *QualityMinute) SetErrNetwork(v int64)

func (*QualityMinute) SetGen 

func (q *QualityMinute) SetGen(g uint64)

func (*QualityMinute) SetHist 

func (q *QualityMinute) SetHist(h [10]int64)

func (*QualityMinute) SetImages 

func (q *QualityMinute) SetImages(v int64)

func (*QualityMinute) SetInputTokens 

func (q *QualityMinute) SetInputTokens(v int64)

func (*QualityMinute) SetOutputTokens 

func (q *QualityMinute) SetOutputTokens(v int64)

func (*QualityMinute) SetSuccesses 

func (q *QualityMinute) SetSuccesses(v int64)

func (*QualityMinute) SetSumQ32 

func (q *QualityMinute) SetSumQ32(v int64)

func (*QualityMinute) SetSumSqQ32 

func (q *QualityMinute) SetSumSqQ32(v int64)

func (*QualityMinute) SetTTFTCount 

func (q *QualityMinute) SetTTFTCount(v int64)

func (*QualityMinute) Successes 

func (q *QualityMinute) Successes() int64

func (*QualityMinute) SumQ32 

func (q *QualityMinute) SumQ32() int64

func (*QualityMinute) SumSqQ32 

func (q *QualityMinute) SumSqQ32() int64

func (*QualityMinute) TTFTCount 

func (q *QualityMinute) TTFTCount() int64

type Recorder 

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

func NewRecorder 

func NewRecorder(effectiveMaxInflight int64) (*Recorder, error)

func (*Recorder) ActiveCount 

func (r *Recorder) ActiveCount() int

func (*Recorder) AddQualityRow 

func (r *Recorder) AddQualityRow(minute int64, key Key) error

func (*Recorder) Begin 

func (r *Recorder) Begin(key Key) *AttemptContext

func (*Recorder) CellStats 

func (r *Recorder) CellStats(key Key) (attempts, successes, ttftCount, tokens, calls, images int64, sumQ32, sumSq int64, hist [10]int64, errClasses [4]int64, ok bool)

func (*Recorder) Close 

func (r *Recorder) Close() error

func (*Recorder) CloseWithContext 

func (r *Recorder) CloseWithContext(ctx context.Context) error

func (*Recorder) EffectiveMaxInflight 

func (r *Recorder) EffectiveMaxInflight() int64

func (*Recorder) EnqueueQualityMinute 

func (r *Recorder) EnqueueQualityMinute(qm *QualityMinute) error

func (*Recorder) FlowMinute 

func (r *Recorder) FlowMinute(minute int64) (*FlowMinute, bool)

func (*Recorder) FlowOverflow 

func (r *Recorder) FlowOverflow() int64

func (*Recorder) FlowOwner 

func (r *Recorder) FlowOwner() *FlowOwner

FlowOwner returns the managed worker exclusively owning this recorder's flow lane: register it (worker.Worker) and publish it (StatsProvider) beside quality-sync; reverse shutdown then closes quality-sync first and the owner second (the refill dependency).

func (*Recorder) GetOrCreateCell 

func (r *Recorder) GetOrCreateCell(key Key) *Cell

func (*Recorder) GlobalInflight 

func (r *Recorder) GlobalInflight() int64

func (*Recorder) InitAttemptContext 

func (r *Recorder) InitAttemptContext(cell *Cell, out *AttemptContext) bool

func (*Recorder) LiveCells 

func (r *Recorder) LiveCells() map[Key]*QualityMinute

LiveCells returns cloned cumulative totals of all active cells (minute=0 marker; zero-attempt cells carry no signal and are skipped). Compile-lane read-only accessor: called from the background routing-compile lane under the recorder mutex — never from the request path. Clones never alias live cells, so the caller may hold the map across cell mutation.

func (*Recorder) MarkEmitted 

func (r *Recorder) MarkEmitted(key Key, delta *QualityMinute)

MarkEmitted records one durably PG-persisted absolute increment for key. Called by the sync worker at its markQualityCommitted choke point (every successful insert path funnels there); best-effort and infallible. Gen comes stamped on the row: a nonzero generation Rotating entry resets the totals (new cell lifetime starts at zero); generation 0 adds to whatever entry exists (external rows predate the gen discipline).

func (*Recorder) MinuteBucketCount 

func (r *Recorder) MinuteBucketCount() int

MinuteBucketCount reports the total distinct pending minute buckets across both lanes (quality under rec.mu, flow under the owner lock — acquired sequentially, never nested).

func (*Recorder) MinuteOverflow 

func (r *Recorder) MinuteOverflow() int64

func (*Recorder) NewAttemptContext 

func (r *Recorder) NewAttemptContext(cell *Cell) *AttemptContext

func (*Recorder) PendingBytes 

func (r *Recorder) PendingBytes() int64

func (*Recorder) PinnedGauge 

func (r *Recorder) PinnedGauge() int64

func (*Recorder) PinnedRetiredCount 

func (r *Recorder) PinnedRetiredCount() int

func (*Recorder) QualityMinute 

func (r *Recorder) QualityMinute(minute int64, key Key) (*QualityMinute, bool)

func (*Recorder) QualityOverflow 

func (r *Recorder) QualityOverflow() int64

func (*Recorder) Retire 

func (r *Recorder) Retire(key Key)

func (*Recorder) RetiredCount 

func (r *Recorder) RetiredCount() int

func (*Recorder) Snapshot 

func (r *Recorder) Snapshot() *Snapshot

Snapshot returns a fully cloned both-lane view: the finalized snapshot is deep-copied on every read, live reads compose quality (rec.mu) and the FlowOwner accumulator (drained first) with sequential locking.

func (*Recorder) SweepReclaimLocked 

func (r *Recorder) SweepReclaimLocked()

func (*Recorder) UnflushedMinutes 

func (r *Recorder) UnflushedMinutes(now time.Time) map[int64]map[Key]*QualityMinute

UnflushedMinutes returns absolute per-minute rows not yet in PG: pending rows with their labels plus one current-minute residual per active cell with unflushed contributions. Compile-lane read-only accessor (same cold path discipline as LiveCells); the returned clones never alias recorder state. Nil when nothing is unflushed.

func (*Recorder) UnpinnedRetiredCount 

func (r *Recorder) UnpinnedRetiredCount() int

type RowDataError 

type RowDataError struct {
	Msg   string
	Cause error
}

func (*RowDataError) Error 

func (e *RowDataError) Error() string

func (*RowDataError) Unwrap 

func (e *RowDataError) Unwrap() error

type Snapshot 

type Snapshot struct {
	Quality map[int64]map[Key]*QualityMinute
	Flow    map[int64]*FlowMinute
}

type SyncConfig 

type SyncConfig struct {
	BatchSize   int
	InstanceSrc string
}

type SyncStats 

type SyncStats struct {
	PendingQuality     int    `json:"pending_quality"`
	PendingFlow        int    `json:"pending_flow"`
	PendingBytes       int64  `json:"pending_bytes"`
	LagMs              int64  `json:"lag_ms"`
	DroppedQuality     int64  `json:"dropped_quality"`
	DroppedFlow        int64  `json:"dropped_flow"`
	PoisonDropped      int64  `json:"poison_dropped"`
	BatchQuality       int64  `json:"batch_quality"`
	BatchFlow          int64  `json:"batch_flow"`
	FreshnessMs        int64  `json:"freshness_ms"`
	LastRedisMs        int64  `json:"last_redis_ms"`
	LastPGMs           int64  `json:"last_pg_ms"`
	LastRedisAttemptMs int64  `json:"last_redis_attempt_ms"`
	LastRedisError     string `json:"last_redis_error"`
	RedisErrors        int64  `json:"redis_errors"`
	ResidualQuality    int64  `json:"residual_quality"`
	ResidualFlow       int64  `json:"residual_flow"`
}

type SyncWorker 

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

func NewSyncWorker 

func NewSyncWorker(rec *Recorder, rdb *redis.Client, pg PGQualityWriter, cfg SyncConfig, log *logx.Logger, onPersisted func()) *SyncWorker

func (*SyncWorker) Close 

func (w *SyncWorker) Close(ctx context.Context) error

func (*SyncWorker) Name 

func (w *SyncWorker) Name() string

func (*SyncWorker) SetClock 

func (w *SyncWorker) SetClock(fn func() time.Time)

func (*SyncWorker) Start 

func (w *SyncWorker) Start(ctx context.Context) error

func (*SyncWorker) Stats 

func (w *SyncWorker) Stats() any

Stats 满足 handler.StatsProvider 契约（any 直出 JSON，ops 端点零转换）。

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.