From bb28f4f4e3e8214d8f7edcb26f73f69cb3b6c129 Mon Sep 17 00:00:00 2001 From: flyworker Date: Fri, 4 Sep 2026 16:07:15 +0000 Subject: [PATCH] feat: persist request history so Transactions survives a restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Transactions view read a 1000-entry in-memory ring, so every restart emptied it. This node restarts often enough that the earnings panel routinely reports counter resets, and each one wiped the list — which also meant the source filter and pager added in #118 could only ever page through traffic since the last restart. Each served request is now written to a request_history table beside the existing metrics_history, and /inference/requests reads from it. The in-memory ring stays as the fallback and as what the model detail view reads. Recording must never slow down serving, so writes are queued and batched by a background writer rather than made per request — the database allows a single open connection, and a synchronous insert would serialise inference behind SQLite. A queue full enough to overflow drops records and counts them instead of blocking, and shutdown drains what is queued. One row per request, so retention has two limits: RetentionDays for how far back the data stays useful, and MaxRows so a burst cannot fill the disk before the age limit applies. Error reasons are truncated for the same reason — an upstream body is unbounded and there is one per row. Rows written before the source column carry none, and all of them arrived over the WebSocket, so filtering to Hub includes them. Without that an operator's older history would vanish the moment they touched the filter, which is what the in-memory path already does. Also ignores *.db and its WAL sidecars: the database normally lives in $CP_PATH, but CP_PATH can be pointed at the repo during development. --- .gitignore | 9 + conf/config.go | 10 + docs/configuration.md | 21 ++ internal/computing/inference_metrics.go | 16 + internal/computing/inference_service.go | 36 ++ internal/computing/request_history_store.go | 338 ++++++++++++++++++ .../computing/request_history_store_test.go | 234 ++++++++++++ 7 files changed, 664 insertions(+) create mode 100644 internal/computing/request_history_store.go create mode 100644 internal/computing/request_history_store_test.go diff --git a/.gitignore b/.gitignore index bbe5f099..ee00ae9b 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,12 @@ logs/ # Dashboard UI internal/dashboard/ui/node_modules/cp.log + +# SQLite data. The provider's database normally lives in $CP_PATH, outside the +# repo — but CP_PATH can be pointed here during development, and the WAL and +# shared-memory sidecars appear beside it. +*.db +*.db-wal +*.db-shm +*.sqlite +*.sqlite3 diff --git a/conf/config.go b/conf/config.go index bed9fb03..43dc59ca 100644 --- a/conf/config.go +++ b/conf/config.go @@ -86,6 +86,16 @@ type ComputeNode struct { HealthCheck HealthCheck `toml:"HealthCheck,omitempty"` RequestLimits RequestLimits `toml:"RequestLimits,omitempty"` Dashboard Dashboard `toml:"Dashboard,omitempty"` + RequestLog RequestLog `toml:"RequestLog,omitempty"` +} + +// RequestLog controls how long the per-request history kept for the +// Transactions view is retained. It is one row per served request, so both +// limits exist: age for how far back it is useful, and a row cap so a burst +// cannot fill the disk before the age limit bites. +type RequestLog struct { + RetentionDays int `toml:"RetentionDays"` // Default: 7 + MaxRows int64 `toml:"MaxRows"` // Default: 200000 } // Dashboard configures the web UI served by `computing-provider dashboard`. diff --git a/docs/configuration.md b/docs/configuration.md index ff1744d4..b1289d7f 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -298,6 +298,27 @@ overloaded GPU. Both can also be changed at runtime, globally or per model, through the REST API — see the endpoint table in the [main README](../README.md#rest-api). +### Request history + +```toml +[RequestLog] +RetentionDays = 7 # how far back the Transactions view reaches (default 7) +MaxRows = 200000 # hard cap on stored rows (default 200000) +``` + +Every served request is stored, so the Transactions view — and its model and +source filters — survive a restart. Before this existed the list lived only in +a 1000-entry in-memory ring, so each restart emptied it, and a node that +restarts often kept almost no history at all. + +Two limits rather than one: `RetentionDays` is how far back the data stays +useful, and `MaxRows` stops a traffic burst filling the disk before the age +limit ever applies. Whichever binds first wins, and pruning runs hourly. + +Writes are batched and happen off the request path, so recording never delays +serving; under a burst large enough to fill the queue, records are dropped and +the count is logged rather than blocking inference. + ### models.json field reference Each key in `models.json` is the marketplace model ID (must match a value in `Models` above). diff --git a/internal/computing/inference_metrics.go b/internal/computing/inference_metrics.go index 0e962612..ece65939 100644 --- a/internal/computing/inference_metrics.go +++ b/internal/computing/inference_metrics.go @@ -57,6 +57,8 @@ type InferenceMetrics struct { // Request history (circular buffer) requestHistory []RequestMetric + // requestSink persists each request, when a store is attached. + requestSink func(RequestMetric) historyMu sync.RWMutex maxHistorySize int } @@ -407,11 +409,25 @@ func (m *InferenceMetrics) GetPrometheusMetrics() string { return sb.String() } +// SetRequestSink installs a persistent store fed alongside the in-memory ring. +// The ring stays: it answers without touching the database, and it is what the +// model detail view reads. +func (m *InferenceMetrics) SetRequestSink(sink func(RequestMetric)) { + m.historyMu.Lock() + defer m.historyMu.Unlock() + m.requestSink = sink +} + // RecordRequest adds a request to the history circular buffer func (m *InferenceMetrics) RecordRequest(req RequestMetric) { m.historyMu.Lock() defer m.historyMu.Unlock() + if m.requestSink != nil { + // Never blocks — the sink queues and returns. + m.requestSink(req) + } + if len(m.requestHistory) >= m.maxHistorySize { // Remove oldest entry (circular buffer) m.requestHistory = m.requestHistory[1:] diff --git a/internal/computing/inference_service.go b/internal/computing/inference_service.go index 78c29034..967969d8 100644 --- a/internal/computing/inference_service.go +++ b/internal/computing/inference_service.go @@ -70,6 +70,7 @@ type InferenceService struct { retryPolicy *RetryPolicy gpuCollector *GPUMetricsCollector metricsHistory *MetricsHistory + requestStore *RequestStore alertMonitor *alertMonitor selfCheck *selfCheckRunner noticeLimiter *noticeLimiter @@ -116,6 +117,17 @@ func NewInferenceService(nodeID, cpPath string) *InferenceService { // Create metrics history for persistence metricsHistory := NewMetricsHistory() + // Per-request history, so the Transactions view survives a restart. + // GetConfig is nil until InitConfig has run, which is not guaranteed for + // callers that construct the service directly; the store's own defaults + // cover that. + var retentionDays int + var maxRows int64 + if cfg := conf.GetConfig(); cfg != nil { + retentionDays, maxRows = cfg.RequestLog.RetentionDays, cfg.RequestLog.MaxRows + } + requestStore := NewRequestStore(retentionDays, maxRows) + s := &InferenceService{ nodeID: nodeID, cpPath: cpPath, @@ -127,6 +139,7 @@ func NewInferenceService(nodeID, cpPath string) *InferenceService { retryPolicy: retryPolicy, gpuCollector: gpuCollector, metricsHistory: metricsHistory, + requestStore: requestStore, } // Set up registry callbacks to update modelMappings for backward compatibility @@ -345,6 +358,15 @@ func (s *InferenceService) Start() error { s.alertMonitor.Start() s.selfCheck.Start() + // Start persistent request history, and feed it from the metrics recorder. + if s.requestStore != nil { + if err := s.requestStore.Start(); err != nil { + logs.GetLogger().Warnf("Failed to start request history store: %v", err) + } else if s.client != nil && s.client.metrics != nil { + s.client.metrics.SetRequestSink(s.requestStore.Record) + } + } + // Start metrics history recorder if s.metricsHistory != nil { if err := s.metricsHistory.Start(func() *InferenceMetricsData { @@ -384,6 +406,9 @@ func (s *InferenceService) Stop() { if s.metricsHistory != nil { s.metricsHistory.Stop() } + if s.requestStore != nil { + s.requestStore.Stop() + } if s.alertMonitor != nil { s.alertMonitor.Stop() } @@ -1271,7 +1296,18 @@ func (s *InferenceService) GetRequestHistory(limit int, modelFilter string) []Re // QueryRequestHistory returns one page of request history along with the total // matching the filters. +// +// Served from the persistent store when there is one, so the list reaches back +// past the last restart. The in-memory ring is the fallback: it is all that +// exists before the store has started, or if the database is unavailable. func (s *InferenceService) QueryRequestHistory(q RequestHistoryQuery) RequestHistoryPage { + if s.requestStore != nil { + if page, err := s.requestStore.Query(q); err == nil { + return page + } else { + logs.GetLogger().Warnf("Request history query failed, falling back to memory: %v", err) + } + } if s.client == nil { return RequestHistoryPage{Requests: []RequestMetric{}, Limit: q.Limit, Offset: q.Offset} } diff --git a/internal/computing/request_history_store.go b/internal/computing/request_history_store.go new file mode 100644 index 00000000..245795f9 --- /dev/null +++ b/internal/computing/request_history_store.go @@ -0,0 +1,338 @@ +package computing + +import ( + "sync" + "sync/atomic" + "time" + + "github.com/filswan/go-mcs-sdk/mcs/api/common/logs" + "github.com/swanchain/computing-provider-v2/internal/db" + "gorm.io/gorm" +) + +// RequestHistoryEntity is one served request, kept so the Transactions view +// survives a restart. +// +// Until this existed the request list lived only in a 1000-entry in-memory +// ring, so every restart emptied it — and this agent restarts often enough +// that the earnings panel routinely reports counter resets. A filter and a +// pager over a list that only reaches back to the last restart are not much +// use. +type RequestHistoryEntity struct { + ID uint `gorm:"primaryKey;autoIncrement"` + RequestID string `gorm:"index"` + Model string `gorm:"index"` + Source string `gorm:"index"` + StartTime time.Time `gorm:"index;not null"` + EndTime time.Time + LatencyMs float64 + TokensIn int + TokensOut int + Streaming bool + Success bool + // ErrorReason can be an upstream body, so it is truncated before storage: + // a row per request means an unbounded field is an unbounded table. + ErrorReason string +} + +func (RequestHistoryEntity) TableName() string { + return "request_history" +} + +const ( + // maxErrorReasonBytes caps what one failure can cost on disk. + maxErrorReasonBytes = 512 + // writeBatchSize is how many rows one insert carries. The database allows a + // single open connection, so writes are batched rather than made per + // request — otherwise a busy node serialises inference behind SQLite. + writeBatchSize = 200 + // flushInterval bounds how stale the stored history can be. The dashboard + // polls every 10s, so a second or two of lag is invisible. + flushInterval = 2 * time.Second + // queueSize is the burst absorbed before records are dropped. Dropping is + // deliberate: recording history must never slow down or block serving. + queueSize = 4096 +) + +// RequestStore persists request history and serves queries over it. +type RequestStore struct { + queue chan RequestMetric + stop chan struct{} + done chan struct{} + mu sync.Mutex + running bool + + retentionDays int + maxRows int64 + + // dropped counts records discarded because the queue was full, so the + // condition is visible rather than silent. + dropped atomic.Int64 +} + +// NewRequestStore builds a store with the given retention. A non-positive +// value falls back to the default. +func NewRequestStore(retentionDays int, maxRows int64) *RequestStore { + if retentionDays <= 0 { + retentionDays = 7 + } + if maxRows <= 0 { + maxRows = 200_000 + } + return &RequestStore{ + queue: make(chan RequestMetric, queueSize), + stop: make(chan struct{}), + done: make(chan struct{}), + retentionDays: retentionDays, + maxRows: maxRows, + } +} + +// Start migrates the table and begins the writer and pruner. +func (s *RequestStore) Start() error { + s.mu.Lock() + if s.running { + s.mu.Unlock() + return nil + } + s.running = true + s.mu.Unlock() + + database := db.NewDbService() + if database == nil { + return nil + } + if err := database.AutoMigrate(&RequestHistoryEntity{}); err != nil { + return err + } + + go s.writeLoop() + go s.pruneLoop() + logs.GetLogger().Info("Request history store started") + return nil +} + +// Stop drains what is queued and shuts the writer down. +func (s *RequestStore) Stop() { + s.mu.Lock() + if !s.running { + s.mu.Unlock() + return + } + s.running = false + s.mu.Unlock() + + close(s.stop) + <-s.done + if dropped := s.dropped.Load(); dropped > 0 { + logs.GetLogger().Warnf("Request history dropped %d record(s) under load", dropped) + } +} + +// Record queues one request. It never blocks: a full queue drops the record +// rather than making the inference path wait on storage. +func (s *RequestStore) Record(req RequestMetric) { + if s == nil { + return + } + if len(req.ErrorReason) > maxErrorReasonBytes { + req.ErrorReason = req.ErrorReason[:maxErrorReasonBytes] + } + select { + case s.queue <- req: + default: + s.dropped.Add(1) + } +} + +func (s *RequestStore) writeLoop() { + defer close(s.done) + ticker := time.NewTicker(flushInterval) + defer ticker.Stop() + + batch := make([]RequestHistoryEntity, 0, writeBatchSize) + flush := func() { + if len(batch) == 0 { + return + } + if database := db.NewDbService(); database != nil { + if err := database.CreateInBatches(batch, writeBatchSize).Error; err != nil { + logs.GetLogger().Warnf("Failed to persist request history: %v", err) + } + } + batch = batch[:0] + } + + for { + select { + case req := <-s.queue: + batch = append(batch, entityFor(req)) + if len(batch) >= writeBatchSize { + flush() + } + case <-ticker.C: + flush() + case <-s.stop: + // Drain whatever is queued so a clean shutdown does not discard + // the requests served just before it. + for { + select { + case req := <-s.queue: + batch = append(batch, entityFor(req)) + if len(batch) >= writeBatchSize { + flush() + } + continue + default: + } + break + } + flush() + return + } + } +} + +func entityFor(req RequestMetric) RequestHistoryEntity { + return RequestHistoryEntity{ + RequestID: req.RequestID, + Model: req.Model, + Source: string(req.Source), + StartTime: req.StartTime, + EndTime: req.EndTime, + LatencyMs: req.LatencyMs, + TokensIn: req.TokensIn, + TokensOut: req.TokensOut, + Streaming: req.Streaming, + Success: req.Success, + ErrorReason: req.ErrorReason, + } +} + +func (s *RequestStore) pruneLoop() { + // Prune on start as well as on the timer, so a node restarted more often + // than the interval still enforces its retention. + s.prune() + ticker := time.NewTicker(time.Hour) + defer ticker.Stop() + for { + select { + case <-ticker.C: + s.prune() + case <-s.stop: + return + } + } +} + +// prune enforces both limits: age, and a row cap so a burst cannot fill the +// disk before the age limit ever bites. +func (s *RequestStore) prune() { + database := db.NewDbService() + if database == nil { + return + } + + cutoff := time.Now().AddDate(0, 0, -s.retentionDays) + if result := database.Where("start_time < ?", cutoff).Delete(&RequestHistoryEntity{}); result.Error != nil { + logs.GetLogger().Warnf("Failed to prune request history by age: %v", result.Error) + } else if result.RowsAffected > 0 { + logs.GetLogger().Debugf("Pruned %d request history row(s) older than %d days", result.RowsAffected, s.retentionDays) + } + + var count int64 + if err := database.Model(&RequestHistoryEntity{}).Count(&count).Error; err != nil { + return + } + if count <= s.maxRows { + return + } + // Delete the oldest rows above the cap, identified by a threshold id rather + // than by offset: ids are monotonic, so this is one indexed delete instead + // of a scan. + var threshold uint + if err := database.Model(&RequestHistoryEntity{}). + Order("id desc").Offset(int(s.maxRows)).Limit(1). + Pluck("id", &threshold).Error; err != nil || threshold == 0 { + return + } + if result := database.Where("id <= ?", threshold).Delete(&RequestHistoryEntity{}); result.Error != nil { + logs.GetLogger().Warnf("Failed to prune request history by row cap: %v", result.Error) + } else if result.RowsAffected > 0 { + logs.GetLogger().Debugf("Pruned %d request history row(s) over the %d row cap", result.RowsAffected, s.maxRows) + } +} + +// Query returns one page of stored history, newest first, with the total +// matching the filters. +func (s *RequestStore) Query(q RequestHistoryQuery) (RequestHistoryPage, error) { + page := RequestHistoryPage{Requests: []RequestMetric{}, Limit: q.Limit, Offset: q.Offset} + if page.Limit <= 0 { + page.Limit = 100 + } + if page.Offset < 0 { + page.Offset = 0 + } + + database := db.NewDbService() + if database == nil { + return page, nil + } + + build := func() *gorm.DB { + tx := database.Model(&RequestHistoryEntity{}) + if q.Model != "" { + tx = tx.Where("model = ?", q.Model) + } + if q.Source != "" { + if q.Source == string(SourceHub) { + // Rows written before the source field existed carry none, and + // every one of them arrived over the WebSocket. Excluding them + // would make an operator's older history vanish the moment + // they filter to Hub. + tx = tx.Where("source = ? OR source = ''", q.Source) + } else { + tx = tx.Where("source = ?", q.Source) + } + } + return tx + } + + var total int64 + if err := build().Count(&total).Error; err != nil { + return page, err + } + page.Total = int(total) + + var rows []RequestHistoryEntity + if err := build(). + Order("start_time desc, id desc"). + Offset(page.Offset).Limit(page.Limit). + Find(&rows).Error; err != nil { + return page, err + } + for _, row := range rows { + page.Requests = append(page.Requests, RequestMetric{ + RequestID: row.RequestID, + Model: row.Model, + StartTime: row.StartTime, + EndTime: row.EndTime, + LatencyMs: row.LatencyMs, + TokensIn: row.TokensIn, + TokensOut: row.TokensOut, + Streaming: row.Streaming, + Success: row.Success, + ErrorReason: row.ErrorReason, + Source: RequestSource(row.Source), + }) + } + return page, nil +} + +// Dropped reports how many records were discarded because the queue was full. +func (s *RequestStore) Dropped() int64 { + if s == nil { + return 0 + } + return s.dropped.Load() +} diff --git a/internal/computing/request_history_store_test.go b/internal/computing/request_history_store_test.go new file mode 100644 index 00000000..147c194e --- /dev/null +++ b/internal/computing/request_history_store_test.go @@ -0,0 +1,234 @@ +package computing + +import ( + "testing" + "time" + + "github.com/swanchain/computing-provider-v2/internal/db" +) + +// withStore gives each test its own database file, so these exercise the real +// SQLite path rather than a stand-in. +func withStore(t *testing.T, retentionDays int, maxRows int64) *RequestStore { + t.Helper() + db.InitDb(t.TempDir()) + if db.NewDbService() == nil { + t.Skip("database unavailable") + } + store := NewRequestStore(retentionDays, maxRows) + if err := store.Start(); err != nil { + t.Fatalf("start store: %v", err) + } + t.Cleanup(store.Stop) + return store +} + +// flush pushes queued records to disk by stopping the writer, which drains. +func flush(t *testing.T, store *RequestStore) { + t.Helper() + store.Stop() +} + +func record(store *RequestStore, id, model string, source RequestSource, at time.Time) { + store.Record(RequestMetric{ + RequestID: id, Model: model, Source: source, + StartTime: at, EndTime: at.Add(time.Second), + LatencyMs: 120, TokensIn: 10, TokensOut: 5, Success: true, + }) +} + +func TestRequestStorePersistsAndPages(t *testing.T) { + store := withStore(t, 7, 1000) + base := time.Now().UTC().Add(-time.Hour) + for i := 0; i < 5; i++ { + record(store, string(rune('a'+i)), "m", SourceHub, base.Add(time.Duration(i)*time.Minute)) + } + flush(t, store) + + page, err := store.Query(RequestHistoryQuery{Limit: 2}) + if err != nil { + t.Fatalf("query: %v", err) + } + if page.Total != 5 { + t.Errorf("total = %d, want 5", page.Total) + } + if len(page.Requests) != 2 { + t.Fatalf("page held %d rows, want 2", len(page.Requests)) + } + // Newest first. + if page.Requests[0].RequestID != "e" || page.Requests[1].RequestID != "d" { + t.Errorf("page 1 = %s,%s — want the two newest (e,d)", page.Requests[0].RequestID, page.Requests[1].RequestID) + } + + second, err := store.Query(RequestHistoryQuery{Limit: 2, Offset: 2}) + if err != nil { + t.Fatalf("query: %v", err) + } + if second.Requests[0].RequestID != "c" { + t.Errorf("page 2 started at %s, want c", second.Requests[0].RequestID) + } +} + +func TestRequestStoreFiltersByModelAndSource(t *testing.T) { + store := withStore(t, 7, 1000) + now := time.Now().UTC() + record(store, "h1", "alpha", SourceHub, now) + record(store, "p1", "alpha", SourceHealth, now) + record(store, "h2", "beta", SourceHub, now) + flush(t, store) + + hub, err := store.Query(RequestHistoryQuery{Source: string(SourceHub)}) + if err != nil { + t.Fatalf("query: %v", err) + } + if hub.Total != 2 { + t.Errorf("hub total = %d, want 2", hub.Total) + } + + combined, err := store.Query(RequestHistoryQuery{Model: "alpha", Source: string(SourceHub)}) + if err != nil { + t.Fatalf("query: %v", err) + } + if combined.Total != 1 || combined.Requests[0].RequestID != "h1" { + t.Errorf("model+source = %+v, want just h1", combined.Requests) + } +} + +// Rows written before the source column carry none, and all of them arrived +// over the WebSocket. Filtering to Hub must include them, matching what the +// in-memory path does. +func TestRequestStoreTreatsSourcelessRowsAsHub(t *testing.T) { + store := withStore(t, 7, 1000) + now := time.Now().UTC() + store.Record(RequestMetric{RequestID: "old", Model: "m", StartTime: now, Success: true}) // no Source + record(store, "new", "m", SourceHub, now.Add(time.Minute)) + record(store, "probe", "m", SourceHealth, now.Add(2*time.Minute)) + flush(t, store) + + hub, err := store.Query(RequestHistoryQuery{Source: string(SourceHub)}) + if err != nil { + t.Fatalf("query: %v", err) + } + if hub.Total != 2 { + t.Fatalf("hub total = %d, want 2 — the unlabelled row must count as hub", hub.Total) + } +} + +// History outliving a restart is the whole point: a second store over the same +// database must see what the first one wrote. +func TestRequestStoreSurvivesRestart(t *testing.T) { + dir := t.TempDir() + db.InitDb(dir) + if db.NewDbService() == nil { + t.Skip("database unavailable") + } + + first := NewRequestStore(7, 1000) + if err := first.Start(); err != nil { + t.Fatalf("start: %v", err) + } + record(first, "before-restart", "m", SourceHub, time.Now().UTC()) + first.Stop() // drains + + second := NewRequestStore(7, 1000) + if err := second.Start(); err != nil { + t.Fatalf("restart: %v", err) + } + defer second.Stop() + + page, err := second.Query(RequestHistoryQuery{Limit: 10}) + if err != nil { + t.Fatalf("query: %v", err) + } + if page.Total != 1 || page.Requests[0].RequestID != "before-restart" { + t.Errorf("after restart the store held %+v, want the record written before it", page.Requests) + } +} + +func TestRequestStorePrunesByAge(t *testing.T) { + store := withStore(t, 1, 1000) + now := time.Now().UTC() + record(store, "old", "m", SourceHub, now.AddDate(0, 0, -3)) + record(store, "fresh", "m", SourceHub, now) + flush(t, store) + + store.prune() + + page, err := store.Query(RequestHistoryQuery{Limit: 10}) + if err != nil { + t.Fatalf("query: %v", err) + } + if page.Total != 1 || page.Requests[0].RequestID != "fresh" { + t.Errorf("after pruning: %+v, want only the row inside the retention window", page.Requests) + } +} + +func TestRequestStorePrunesByRowCap(t *testing.T) { + store := withStore(t, 30, 3) + now := time.Now().UTC() + for i := 0; i < 10; i++ { + record(store, string(rune('a'+i)), "m", SourceHub, now.Add(time.Duration(i)*time.Second)) + } + flush(t, store) + + store.prune() + + page, err := store.Query(RequestHistoryQuery{Limit: 50}) + if err != nil { + t.Fatalf("query: %v", err) + } + if page.Total != 3 { + t.Errorf("total = %d, want the 3-row cap enforced", page.Total) + } + // The cap must keep the newest, not an arbitrary three. + if page.Requests[0].RequestID != "j" { + t.Errorf("newest kept = %s, want j", page.Requests[0].RequestID) + } +} + +// Recording must never block serving, even when the writer cannot keep up. +func TestRequestStoreRecordNeverBlocks(t *testing.T) { + store := NewRequestStore(7, 1000) // not started: nothing drains the queue + now := time.Now().UTC() + done := make(chan struct{}) + go func() { + for i := 0; i < queueSize+500; i++ { + store.Record(RequestMetric{RequestID: "x", Model: "m", StartTime: now}) + } + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Record blocked when the queue was full — it must drop instead") + } + if store.Dropped() == 0 { + t.Error("overflow should be counted, not silent") + } +} + +// An upstream error body can be arbitrarily long, and there is one row per +// request. +func TestRequestStoreTruncatesLongErrors(t *testing.T) { + store := withStore(t, 7, 1000) + long := make([]byte, 4000) + for i := range long { + long[i] = 'x' + } + store.Record(RequestMetric{ + RequestID: "e", Model: "m", Source: SourceHub, + StartTime: time.Now().UTC(), ErrorReason: string(long), + }) + flush(t, store) + + page, err := store.Query(RequestHistoryQuery{Limit: 1}) + if err != nil { + t.Fatalf("query: %v", err) + } + if len(page.Requests) != 1 { + t.Fatalf("expected the row back") + } + if got := len(page.Requests[0].ErrorReason); got != maxErrorReasonBytes { + t.Errorf("stored error reason is %d bytes, want it capped at %d", got, maxErrorReasonBytes) + } +}