diff --git a/internal/hashlogcompare/metrics.go b/internal/hashlogcompare/metrics.go index 95d27740..229e24aa 100644 --- a/internal/hashlogcompare/metrics.go +++ b/internal/hashlogcompare/metrics.go @@ -13,6 +13,7 @@ const ( ReasonNotFound = "not_found" ReasonUnavailable = "unavailable" ReasonCoverageGap = "coverage_gap" + ReasonTornRow = "torn_row" ReasonInvalidRow = "invalid_row" ) @@ -103,7 +104,7 @@ func (m *Metrics) initPair(chain, migrating, reserve string, now time.Time) { m.ComparedHeights.WithLabelValues(chain, migrating, reserve) for _, role := range []string{RoleMigrating, RoleReserve} { m.HeightGaps.WithLabelValues(chain, migrating, reserve, role) - for _, reason := range []string{ReasonNotFound, ReasonUnavailable, ReasonCoverageGap, ReasonInvalidRow} { + for _, reason := range []string{ReasonNotFound, ReasonUnavailable, ReasonCoverageGap, ReasonTornRow, ReasonInvalidRow} { m.SourceErrors.WithLabelValues(chain, migrating, reserve, role, reason) } } diff --git a/internal/hashlogcompare/reader.go b/internal/hashlogcompare/reader.go index 148e8a1a..bda691ec 100644 --- a/internal/hashlogcompare/reader.go +++ b/internal/hashlogcompare/reader.go @@ -25,6 +25,10 @@ const ( // while it was being read), so the reader restarts at the tip. var errCoverageGap = errors.New("hash log coverage gap") +// errTornRow means a sealed file ends in an incomplete line: a row that a +// crash tore mid-write. The reader drops it and moves to the next file. +var errTornRow = errors.New("hash log torn row") + // Source is the subset of the sidecar client the reader uses. type Source interface { ListHashLog(ctx context.Context) ([]sidecar.HashLogFile, error) @@ -40,7 +44,11 @@ type Row struct { // Reader tails one node's hash log in file-index order. It keeps a byte // cursor into the current file and only consumes complete lines, so a line -// that is half written when read is picked up whole on the next poll. +// that is half written when read is picked up whole on the next poll. A +// sealed file gets no more bytes, so the reader drops an incomplete last +// line there, as the HashLogger's own reader does. The HashLogger seals only +// a file with a complete header and at least one complete row, so that last +// line is the only incomplete line a sealed file can hold. type Reader struct { src Source @@ -60,7 +68,8 @@ func NewReader(src Source) *Reader { // Poll returns the complete rows written since the last call, plus the number // of lines that could not be parsed. Transport errors leave the cursor where -// it was; errCoverageGap resets it to the tip. +// it was; errCoverageGap resets it to the tip; errTornRow moves it past the +// dropped line. func (r *Reader) Poll(ctx context.Context) (rows []Row, invalid int, err error) { if !r.started { ok, err := r.start(ctx) @@ -111,9 +120,13 @@ func (r *Reader) Poll(ctx context.Context) (rows []Row, invalid int, err error) rows = append(rows, parsed...) invalid += bad - if !r.sealed || len(complete) < len(data) { + if !r.sealed { return rows, invalid, nil } + if torn := len(data) - len(complete); torn > 0 { + r.offset += int64(torn) + return rows, invalid, fmt.Errorf("%w: dropped %d trailing bytes of sealed file %s", errTornRow, torn, r.name) + } more, err := r.nextFile(ctx) if err != nil || !more { return rows, invalid, err diff --git a/internal/hashlogcompare/reader_test.go b/internal/hashlogcompare/reader_test.go index 0f0c4844..703d56cc 100644 --- a/internal/hashlogcompare/reader_test.go +++ b/internal/hashlogcompare/reader_test.go @@ -83,6 +83,27 @@ func TestReader_FollowsSealAndNextFile(t *testing.T) { g.Expect(got).To(BeEmpty()) } +func TestReader_DropsTornTailOfSealedFile(t *testing.T) { + g := NewWithT(t) + src := &fakeSource{} + f := src.add(1, testHeader+rows(10, 11)) + r := NewReader(src) + + got, _, err := r.Poll(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(heights(got)).To(Equal(span(10, 11))) + + f.seal(10, 12, row(12, "")+row(13, "")[:5]) + src.add(2, testHeader+rows(13, 14)) + got, _, err = r.Poll(t.Context()) + g.Expect(err).To(MatchError(errTornRow)) + g.Expect(heights(got)).To(Equal([]int64{12})) + + got, _, err = r.Poll(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(heights(got)).To(Equal(span(13, 14))) +} + func TestReader_SkipsRemovedEmptyFile(t *testing.T) { g := NewWithT(t) src := &fakeSource{} diff --git a/internal/hashlogcompare/run.go b/internal/hashlogcompare/run.go index a6568e54..29e3c8d9 100644 --- a/internal/hashlogcompare/run.go +++ b/internal/hashlogcompare/run.go @@ -87,6 +87,8 @@ func (pr *pairRunner) read(ctx context.Context, role string, r *Reader, buf map[ reason = ReasonNotFound case errors.Is(err, errCoverageGap): reason = ReasonCoverageGap + case errors.Is(err, errTornRow): + reason = ReasonTornRow } m.SourceErrors.WithLabelValues(append(roleLabels, reason)...).Inc() pr.c.Log.Warn("hash log read failed", pr.logAttrs(role, "reason", reason, "error", err)...) diff --git a/internal/hashlogcompare/run_test.go b/internal/hashlogcompare/run_test.go index 01b01653..ba60585b 100644 --- a/internal/hashlogcompare/run_test.go +++ b/internal/hashlogcompare/run_test.go @@ -107,6 +107,28 @@ func TestPairRunner_MissingReserveLogIsSourceError(t *testing.T) { g.Expect(value(t, m.ComparedHeights.WithLabelValues(pr.labels...))).To(BeZero()) } +func TestPairRunner_TornReserveRowIsSourceError(t *testing.T) { + g := NewWithT(t) + migrating, reserve := &fakeSource{}, &fakeSource{} + migrating.add(1, testHeader+rows(1, 6)) + rf := reserve.add(1, testHeader+rows(1, 3)+row(4, "")[:5]) + pr, m := newTestRunner(migrating, reserve) + reserveTorn := m.SourceErrors.WithLabelValues(append(pr.labels, RoleReserve, ReasonTornRow)...) + + pr.poll(t.Context()) + g.Expect(value(t, reserveTorn)).To(BeZero()) + + rf.seal(1, 3, "") + reserve.add(2, testHeader+rows(4, 6)) + pr.poll(t.Context()) + g.Expect(value(t, reserveTorn)).To(Equal(1.0)) + + pr.poll(t.Context()) + g.Expect(value(t, reserveTorn)).To(Equal(1.0)) + g.Expect(value(t, m.LastComparedHeight.WithLabelValues(pr.labels...))).To(Equal(6.0)) + g.Expect(value(t, m.SourceHeight.WithLabelValues(append(pr.labels, RoleReserve)...))).To(Equal(6.0)) +} + func TestInitPair_ExportsZeroSeries(t *testing.T) { g := NewWithT(t) reg := prometheus.NewRegistry() @@ -123,5 +145,5 @@ func TestInitPair_ExportsZeroSeries(t *testing.T) { g.Expect(series).To(HaveKeyWithValue("sei_hashlog_compare_mismatches_total", 1)) g.Expect(series).To(HaveKeyWithValue("sei_hashlog_compare_last_compared_timestamp_seconds", 1)) g.Expect(value(t, m.LastComparedTimestamp.WithLabelValues("c", "a", "b"))).To(Equal(1_800_000_000.0)) - g.Expect(series).To(HaveKeyWithValue("sei_hashlog_compare_source_errors_total", 2*4)) + g.Expect(series).To(HaveKeyWithValue("sei_hashlog_compare_source_errors_total", 2*5)) }