perf(gr26): flush shelter ingest writes once per parquet chunk

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 ───────────────────────────────────────
@@ -97,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) {
@@ -133,87 +139,120 @@ 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
+	chunkIdx := 0
+	var flushErr error
 	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. 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.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.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
 }
 
-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 ───────────────────────────────────────────────────────
 
 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
@@ -226,13 +265,40 @@ 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),
 		errorByCanID:   make(map[int]decodeErrorSample),
 	}
 }
 
+// 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"`
@@ -279,18 +345,45 @@ 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"`
 	Decoded                 int                      `json:"decoded"`
 	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 {

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))
 		}

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