From 1ddd48a9f99bfe28f5caa41964f126f2b6f059a5 Mon Sep 17 00:00:00 2001 From: Bharat Kathi Date: Thu, 6 Aug 2026 20:47:14 -0700 Subject: [PATCH 1/2] perf(gr26): flush shelter ingest writes once per parquet chunk --- gr26/job/ingest_batch.go | 52 +++++++++++++++++++++++----------------- gr26/service/can.go | 27 +++++++++++++++++++++ 2 files changed, 57 insertions(+), 22 deletions(-) diff --git a/gr26/job/ingest_batch.go b/gr26/job/ingest_batch.go index a97626aa..5c717ee1 100644 --- a/gr26/job/ingest_batch.go +++ b/gr26/job/ingest_batch.go @@ -19,9 +19,12 @@ import ( ulid "github.com/gaucho-racing/ulid-go" gr26config "github.com/gaucho-racing/mapache/gr26/config" + "github.com/gaucho-racing/mapache/gr26/model" "github.com/gaucho-racing/mapache/gr26/pkg/foreman" "github.com/gaucho-racing/mapache/gr26/pkg/logger" "github.com/gaucho-racing/mapache/gr26/service" + + mapache "github.com/gaucho-racing/mapache/mapache-go/v3" ) // ─── producer side: gr26.ingest_batch ─────────────────────────────────────── @@ -133,11 +136,27 @@ func processFile(ctx context.Context, client *s3.Client, key string, progress *f stats := newIngestStats() const chunk = 4096 rows := make([]shelterRow, chunk) + cans := make([]model.CAN, 0, chunk) + signals := make([]mapache.Signal, 0, chunk) total := 0 for { n, readErr := pr.Read(rows) + cans, signals = cans[:0], signals[:0] for i := 0; i < n; i++ { - dispatchRow(rows[i], stats) + can, sigs, ok := dispatchRow(rows[i], stats) + if !ok { + continue + } + cans = append(cans, can) + signals = append(signals, sigs...) + } + // One flush per chunk; log-and-continue so a failed insert doesn't + // drop the rest of the file (retries re-cover it via RMT dedup). + if err := service.CreateCANs(cans); err != nil { + logger.SugarLogger.Infof("Error creating CAN records: %s", err) + } + if err := service.CreateSignals(signals); err != nil { + logger.SugarLogger.Infof("Error creating signals: %s", err) } total += n progress.Set(int64(total), totalRows, "decoding parquet rows") @@ -168,36 +187,25 @@ func processFile(ctx context.Context, client *s3.Client, key string, progress *f }, nil } -func dispatchRow(r shelterRow, stats *ingestStats) { +// dispatchRow is the cold-storage decode path — persistence happens in +// per-chunk flushes upstream. UploadKey stays 0 (bucket access is the +// trust boundary), and no WS/side-channel firing — historical data, and +// we don't want to re-enqueue shelter batches. +func dispatchRow(r shelterRow, stats *ingestStats) (model.CAN, []mapache.Signal, bool) { // Topic format: gr26/{vehicle}/{node}/0x{can_id_hex} parts := strings.Split(r.Topic, "/") if len(parts) != 4 { - return + return model.CAN{}, nil, false } nodeID := parts[2] canIDStr := strings.TrimPrefix(parts[3], "0x") canIDInt, err := strconv.ParseInt(canIDStr, 16, 64) if err != nil { - return - } - replayFrame(r.VehicleID, nodeID, int(canIDInt), int(r.Timestamp), r.Data, stats) -} - -// replayFrame is the cold-storage decode-and-persist path. UploadKey -// stays 0 (bucket access is the trust boundary), and no WS/side-channel -// firing — historical data, and we don't want to re-enqueue shelter batches. -func replayFrame(vehicleID, nodeID string, canID, ts int, data []byte, stats *ingestStats) { - can, signals := service.ProcessFrame(vehicleID, nodeID, canID, ts, data) - stats.record(canID, nodeID, ts, can.Metadata) - - if _, err := service.CreateCAN(can); err != nil { - logger.SugarLogger.Infof("Error creating CAN record: %s", err) - } - if len(signals) > 0 { - if err := service.CreateSignals(signals); err != nil { - logger.SugarLogger.Infof("Error creating signals: %s", err) - } + return model.CAN{}, nil, false } + can, signals := service.ProcessFrame(r.VehicleID, nodeID, int(canIDInt), int(r.Timestamp), r.Data) + stats.record(int(canIDInt), nodeID, int(r.Timestamp), can.Metadata) + return can, signals, true } // ─── result reporting ─────────────────────────────────────────────────────── diff --git a/gr26/service/can.go b/gr26/service/can.go index d241cd77..ef7794de 100644 --- a/gr26/service/can.go +++ b/gr26/service/can.go @@ -89,6 +89,7 @@ func GetSignalsForCAN(canMessageID string) ([]mapache.Signal, error) { // Dedup on (vehicle_id, node_id, timestamp) is handled by the // ReplacingMergeTree engine — latest created_at wins on merge. const insertCANSQL = `INSERT INTO gr26_can (id, vehicle_id, node_id, timestamp, can_id, bytes, upload_key, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` +const insertCANBatchSQL = `INSERT INTO gr26_can (id, vehicle_id, node_id, timestamp, can_id, bytes, upload_key, metadata)` func CreateCAN(can model.CAN) (model.CAN, error) { can.ID = ulid.Make().Prefixed("can") @@ -105,6 +106,32 @@ func CreateCAN(can model.CAN) (model.CAN, error) { return can, nil } +// CreateCANs is the bulk-ingest counterpart of CreateCAN: one buffered block +// insert per call — a single round trip regardless of frame count. async_insert +// (InsertCtx) doesn't apply to native block inserts, so plain ctx here. +func CreateCANs(cans []model.CAN) error { + for i := range cans { + cans[i].ID = ulid.Make().Prefixed("can") + } + if !config.ClickhouseEnabled() || len(cans) == 0 { + return nil + } + batch, err := database.Conn.PrepareBatch(context.Background(), insertCANBatchSQL) + if err != nil { + return err + } + defer batch.Close() + for _, c := range cans { + if err := batch.Append( + c.ID, c.VehicleID, c.NodeID, int64(c.Timestamp), int32(c.CANID), + string(c.Bytes), int32(c.UploadKey), string(c.Metadata), + ); err != nil { + return err + } + } + return batch.Send() +} + func scanCANRow(row driver.Row) (model.CAN, error) { var ( id, vehicleID, nodeID string From 0093f217cc887d2a526725f2ba2782499803fb74 Mon Sep 17 00:00:00 2001 From: Bharat Kathi Date: Fri, 7 Aug 2026 08:28:01 -0700 Subject: [PATCH 2/2] fix(gr26): fail shelter ingest job when a chunk insert fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-chunk flush errors were only logged, so processFile returned success and foreman completed the job instead of spending a retry attempt — one transient ClickHouse blip could silently drop up to 4,096 frames. Hold the first flush error and return it after the loop, so the rest of the file still lands but the attempt fails and gets retried. Report stats on the failure path too. FailRequest already carries a result, but the worker only forwarded one to Complete; wire it into Fail so a partial-failure run records failed_inserts, cans_dropped, signals_dropped and per-chunk error samples alongside the decode counts. --- gr26/job/ingest_batch.go | 127 +++++++++++++++++++++++++++++++------ gr26/pkg/foreman/worker.go | 5 ++ 2 files changed, 111 insertions(+), 21 deletions(-) diff --git a/gr26/job/ingest_batch.go b/gr26/job/ingest_batch.go index 5c717ee1..17ea21bd 100644 --- a/gr26/job/ingest_batch.go +++ b/gr26/job/ingest_batch.go @@ -100,12 +100,15 @@ func IngestBatchHandler(ctx context.Context, job foreman.Job, progress *foreman. return nil, err } - res, err := processFile(ctx, client, shelterKey(p.VehicleID, p.FileULID), progress) + // Report stats on both paths — foreman records the result on a failed run + // too, so a partial-failure attempt still says what landed and what didn't. + res, procErr := processFile(ctx, client, shelterKey(p.VehicleID, p.FileULID), progress) + res.FileULID = p.FileULID + payload, err := json.Marshal(res) if err != nil { - return nil, err + return nil, errors.Join(procErr, fmt.Errorf("encode result: %w", err)) } - res.FileULID = p.FileULID - return json.Marshal(res) + return payload, procErr } func processFile(ctx context.Context, client *s3.Client, key string, progress *foreman.Progress) (ingestResult, error) { @@ -139,6 +142,8 @@ func processFile(ctx context.Context, client *s3.Client, key string, progress *f cans := make([]model.CAN, 0, chunk) signals := make([]mapache.Signal, 0, chunk) total := 0 + chunkIdx := 0 + var flushErr error for { n, readErr := pr.Read(rows) cans, signals = cans[:0], signals[:0] @@ -150,41 +155,53 @@ func processFile(ctx context.Context, client *s3.Client, key string, progress *f cans = append(cans, can) signals = append(signals, sigs...) } - // One flush per chunk; log-and-continue so a failed insert doesn't - // drop the rest of the file (retries re-cover it via RMT dedup). + // One flush per chunk. Keep going past a failed insert so the rest of + // the file still lands, but hold onto the first error and fail the job + // below — a dropped chunk is up to 4,096 frames, so it has to burn a + // foreman attempt rather than report success (RMT dedup makes the + // re-processed chunks harmless). if err := service.CreateCANs(cans); err != nil { - logger.SugarLogger.Infof("Error creating CAN records: %s", err) + logger.SugarLogger.Errorf("[SHELTER] %s: chunk %d: failed to insert %d CAN records: %s", key, chunkIdx, len(cans), err) + stats.recordInsertFailure(insertTargetCANs, chunkIdx, len(cans), err) + if flushErr == nil { + flushErr = fmt.Errorf("insert cans: %w", err) + } } if err := service.CreateSignals(signals); err != nil { - logger.SugarLogger.Infof("Error creating signals: %s", err) + logger.SugarLogger.Errorf("[SHELTER] %s: chunk %d: failed to insert %d signals: %s", key, chunkIdx, len(signals), err) + stats.recordInsertFailure(insertTargetSignals, chunkIdx, len(signals), err) + if flushErr == nil { + flushErr = fmt.Errorf("insert signals: %w", err) + } } total += n + chunkIdx++ progress.Set(int64(total), totalRows, "decoding parquet rows") if errors.Is(readErr, io.EOF) { break } if readErr != nil { - return ingestResult{}, fmt.Errorf("parquet read: %w", readErr) + return stats.result(total, time.Since(start)), fmt.Errorf("parquet read: %w", readErr) } } duration := time.Since(start) + if flushErr != nil { + // Every row decoded, so the counters legitimately read 100% — don't + // claim "complete" though, the inserts didn't all land. Best-effort: + // only lands if a heartbeat tick catches it before the handler returns. + progress.Set(int64(total), totalRows, fmt.Sprintf("%d chunk insert(s) failed", stats.failedInserts)) + logger.SugarLogger.Errorf("[SHELTER] %s: %d rows in %s, %d chunk insert(s) failed (cans_dropped=%d signals_dropped=%d)", + key, total, duration, stats.failedInserts, stats.cansDropped, stats.signalsDropped) + return stats.result(total, duration), flushErr + } + // Pin to (total, total) so the final heartbeat stores a clean 100%. progress.Set(totalRows, totalRows, "complete") logger.SugarLogger.Infof("[SHELTER] %s: %d rows in %s (decoded=%d unknown=%d errors=%d invalid_timestamp=%d)", key, total, duration, stats.decoded, stats.unknown, stats.decodeError, stats.invalidTimestamp) - return ingestResult{ - TotalRows: total, - Decoded: stats.decoded, - UnknownCanID: stats.unknown, - DecodeError: stats.decodeError, - InvalidTimestamp: stats.invalidTimestamp, - DurationMs: duration.Milliseconds(), - UnknownBreakdown: topUnknown(stats.unknownByCanID, 10), - DecodeErrorBreakdown: topErrors(stats.errorByCanID, 10), - InvalidTimestampSamples: stats.invalidTimestampSamples, - }, nil + return stats.result(total, duration), nil } // dispatchRow is the cold-storage decode path — persistence happens in @@ -211,17 +228,31 @@ func dispatchRow(r shelterRow, stats *ingestStats) (model.CAN, []mapache.Signal, // ─── result reporting ─────────────────────────────────────────────────────── const invalidTimestampSampleLimit = 10 +const insertFailureSampleLimit = 10 type ingestStats struct { decoded int unknown int decodeError int invalidTimestamp int + failedInserts int + cansDropped int + signalsDropped int unknownByCanID map[int]int errorByCanID map[int]decodeErrorSample invalidTimestampSamples []invalidTimestampSample + insertFailureSamples []insertFailureSample } +// insertTarget names which per-chunk flush failed, and lands verbatim in +// the job result. +type insertTarget string + +const ( + insertTargetCANs insertTarget = "cans" + insertTargetSignals insertTarget = "signals" +) + type decodeErrorSample struct { count int sample string @@ -234,6 +265,13 @@ type invalidTimestampSample struct { NodeID string `json:"node_id"` } +type insertFailureSample struct { + Target insertTarget `json:"target"` + Chunk int `json:"chunk"` + Rows int `json:"rows"` + Error string `json:"error"` +} + func newIngestStats() *ingestStats { return &ingestStats{ unknownByCanID: make(map[int]int), @@ -241,6 +279,26 @@ func newIngestStats() *ingestStats { } } +// recordInsertFailure notes a chunk that decoded fine but never landed in +// ClickHouse, so the result can distinguish "decoded" from "persisted". +func (s *ingestStats) recordInsertFailure(target insertTarget, chunk, rows int, err error) { + s.failedInserts++ + switch target { + case insertTargetCANs: + s.cansDropped += rows + case insertTargetSignals: + s.signalsDropped += rows + } + if len(s.insertFailureSamples) < insertFailureSampleLimit { + s.insertFailureSamples = append(s.insertFailureSamples, insertFailureSample{ + Target: target, + Chunk: chunk, + Rows: rows, + Error: err.Error(), + }) + } +} + func (s *ingestStats) record(canID int, nodeID string, ts int, metadata []byte) { var meta struct { Status string `json:"status"` @@ -287,7 +345,10 @@ type errorBreakdownEntry struct { SampleError string `json:"sample_error,omitempty"` } -// ingestResult lands on foreman.job.result. Top-N caps bound payload size. +// ingestResult lands on foreman.job.result — on failed attempts too, so a +// partial-failure run still reports what landed. Top-N caps bound payload +// size. Decoded counts what decoded, not what persisted: subtract +// CANsDropped / SignalsDropped for that. type ingestResult struct { FileULID string `json:"file_ulid,omitempty"` TotalRows int `json:"total_rows"` @@ -295,10 +356,34 @@ type ingestResult struct { UnknownCanID int `json:"unknown_can_id"` DecodeError int `json:"decode_error"` InvalidTimestamp int `json:"invalid_timestamp,omitempty"` + FailedInserts int `json:"failed_inserts,omitempty"` + CANsDropped int `json:"cans_dropped,omitempty"` + SignalsDropped int `json:"signals_dropped,omitempty"` DurationMs int64 `json:"duration_ms"` UnknownBreakdown []breakdownEntry `json:"unknown_breakdown,omitempty"` DecodeErrorBreakdown []errorBreakdownEntry `json:"decode_error_breakdown,omitempty"` InvalidTimestampSamples []invalidTimestampSample `json:"invalid_timestamp_samples,omitempty"` + InsertFailureSamples []insertFailureSample `json:"insert_failure_samples,omitempty"` +} + +// result snapshots the stats into the foreman result payload. Called on the +// success and failure paths alike so a failed attempt still reports coverage. +func (s *ingestStats) result(totalRows int, duration time.Duration) ingestResult { + return ingestResult{ + TotalRows: totalRows, + Decoded: s.decoded, + UnknownCanID: s.unknown, + DecodeError: s.decodeError, + InvalidTimestamp: s.invalidTimestamp, + FailedInserts: s.failedInserts, + CANsDropped: s.cansDropped, + SignalsDropped: s.signalsDropped, + DurationMs: duration.Milliseconds(), + UnknownBreakdown: topUnknown(s.unknownByCanID, 10), + DecodeErrorBreakdown: topErrors(s.errorByCanID, 10), + InvalidTimestampSamples: s.invalidTimestampSamples, + InsertFailureSamples: s.insertFailureSamples, + } } func topUnknown(m map[int]int, n int) []breakdownEntry { diff --git a/gr26/pkg/foreman/worker.go b/gr26/pkg/foreman/worker.go index 5e3b7e96..0c9c8a63 100644 --- a/gr26/pkg/foreman/worker.go +++ b/gr26/pkg/foreman/worker.go @@ -67,6 +67,10 @@ import ( // Returning an error → Fail. By default it's retryable with the // Worker's DefaultBackoffSec; return a *FailError to override, or // errors.Is(err, ErrPermanent) for a non-retryable terminalization. +// +// Bytes and an error together are valid: the run still fails, but the +// result is recorded on it. Partial-failure handlers should use this to +// report what they did land before giving up. type Handler func(ctx context.Context, job Job, progress *Progress) (json.RawMessage, error) // ErrPermanent marks a handler error as non-retryable. The job @@ -308,6 +312,7 @@ func (w *Worker) handleOne(parent context.Context, claimed Claimed, onErr func(e Error: msg, Retryable: retryable, BackoffSec: backoff, + Result: result, }); err != nil { onErr(fmt.Errorf("fail: %w", err)) }