Skip to content

Commit 3448637

Browse files
frahlgclaude
andauthored
fix(state): one unreadable Parquet file never stops boot (#1901)
* fix(state): one unreadable Parquet file never stops boot AbsorbColdHistory returned the first read error, and main.go exits on it. The bad file stayed in place, so every restart failed the same way. Now an unreadable bucket file is logged, kept for recovery and listed as unreadable_cold_files in the history health report. The other days import, and the day's older legacy copy is used when it reads. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(state): move an unreadable bucket file out of the read path The first version kept a damaged DD.buckets.parquet at its normal name. aggregatePaths still found it, so every series query over that day failed with the Parquet error, on every boot. Rename the file to DD.buckets.parquet.unreadable instead. No reader, importer or converter opens a name that does not end in .parquet. List these files in history health on every boot. Keep the day's legacy copy on disk, unimported, while its original is unreadable. It serves reads for the day. Importing it would make a good copy of the original, restored later, conflict with the stored hours and stop boot with "archive differs from stored hour". Tests open the store with the same cold folder for reads and absorb, as main.go does, and cover a restored original. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent fd928c6 commit 3448637

5 files changed

Lines changed: 270 additions & 5 deletions

File tree

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"ftw": patch
3+
---
4+
5+
Start FTW when one old history file cannot be read. FTW renames the file with an `.unreadable` suffix and keeps it, imports the other days, keeps that day's older copy for history views, and lists the file under `unreadable_cold_files` in the history health report.
Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
package state
2+
3+
import (
4+
"context"
5+
"os"
6+
"path/filepath"
7+
"slices"
8+
"testing"
9+
"time"
10+
)
11+
12+
func unreadableBucket(at time.Time, v float64) metricBucket {
13+
return metricBucket{Driver: "meter", Metric: "grid_w", BucketSummary: BucketSummary{StartMS: at.UnixMilli(), ResolutionMS: ArchiveResolutionMS, FirstMS: at.UnixMilli(), LastMS: at.UnixMilli() + 1000, N: 1, Sum: v, Min: v, Max: v, Last: v}}
14+
}
15+
16+
// unreadableColdFixture opens a store the way main.go does, with the same
17+
// cold folder for reads and for the boot import, so runtime readers see
18+
// whatever the import leaves behind.
19+
func unreadableColdFixture(t *testing.T) (*Store, string, time.Time) {
20+
t.Helper()
21+
s := freshStore(t)
22+
if err := s.EnableHistoryAggregation(); err != nil {
23+
t.Fatal(err)
24+
}
25+
cold := t.TempDir()
26+
s.coldDir = cold
27+
return s, cold, time.Now().UTC().Add(-40 * 24 * time.Hour).Truncate(24 * time.Hour)
28+
}
29+
30+
func dayFile(t *testing.T, cold string, day time.Time, suffix string) string {
31+
t.Helper()
32+
path := filepath.Join(cold, day.Format("2006/01"), day.Format("02")+suffix)
33+
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
34+
t.Fatal(err)
35+
}
36+
return path
37+
}
38+
39+
func assertFile(t *testing.T, path string, want bool) {
40+
t.Helper()
41+
_, err := os.Stat(path)
42+
if want && err != nil {
43+
t.Fatalf("%s must stay: %v", path, err)
44+
}
45+
if !want && !os.IsNotExist(err) {
46+
t.Fatalf("%s must be gone: %v", path, err)
47+
}
48+
}
49+
50+
func listedUnreadable(s *Store) []string {
51+
listed, _ := s.HistoryBackend()["unreadable_cold_files"].([]string)
52+
return listed
53+
}
54+
55+
// One damaged bucket day must not stop boot-time absorb of the other days.
56+
// main.go exits on an AbsorbColdHistory error, and the file would stay, so
57+
// every restart would fail the same way. The kept file must also leave the
58+
// day's history reads working.
59+
func TestAbsorbColdHistorySkipsUnreadableBucketFile(t *testing.T) {
60+
damage := map[string]func(t *testing.T, path string){
61+
"truncated": func(t *testing.T, path string) {
62+
info, err := os.Stat(path)
63+
if err != nil {
64+
t.Fatal(err)
65+
}
66+
if err := os.Truncate(path, info.Size()/2); err != nil {
67+
t.Fatal(err)
68+
}
69+
},
70+
"no read permission": func(t *testing.T, path string) {
71+
if os.Geteuid() == 0 {
72+
t.Skip("root reads files without permission")
73+
}
74+
if err := os.Chmod(path, 0); err != nil {
75+
t.Fatal(err)
76+
}
77+
t.Cleanup(func() { _ = os.Chmod(path, 0o644); _ = os.Chmod(path+unreadableColdSuffix, 0o644) })
78+
},
79+
}
80+
for name, spoil := range damage {
81+
t.Run(name, func(t *testing.T) {
82+
s, cold, good := unreadableColdFixture(t)
83+
bad := good.Add(24 * time.Hour)
84+
hour := good.Add(3 * time.Hour)
85+
badHour := bad.Add(5 * time.Hour)
86+
goodPath := dayFile(t, cold, good, ".buckets.parquet")
87+
badPath := dayFile(t, cold, bad, ".buckets.parquet")
88+
legacyPath := dayFile(t, cold, bad, ".legacy-buckets.parquet")
89+
if err := writeTestBuckets(goodPath, []metricBucket{unreadableBucket(hour, 10)}); err != nil {
90+
t.Fatal(err)
91+
}
92+
if err := writeTestBuckets(badPath, []metricBucket{unreadableBucket(bad, 5)}); err != nil {
93+
t.Fatal(err)
94+
}
95+
spoil(t, badPath)
96+
// The older copy of the damaged day is readable and serves it.
97+
if err := writeTestBuckets(legacyPath, []metricBucket{unreadableBucket(badHour, 7)}); err != nil {
98+
t.Fatal(err)
99+
}
100+
101+
for boot := 1; boot <= 2; boot++ {
102+
if err := s.AbsorbColdHistory(context.Background(), cold); err != nil {
103+
t.Fatalf("boot %d: AbsorbColdHistory: %v", boot, err)
104+
}
105+
assertFile(t, goodPath, false)
106+
assertFile(t, badPath, false)
107+
assertFile(t, badPath+unreadableColdSuffix, true)
108+
assertFile(t, legacyPath, true)
109+
if want := bad.Format("2006/01/02") + ".buckets.parquet" + unreadableColdSuffix; !slices.Equal(listedUnreadable(s), []string{want}) {
110+
t.Fatalf("boot %d: unreadable_cold_files = %v, want [%s]", boot, listedUnreadable(s), want)
111+
}
112+
for _, want := range []struct {
113+
at time.Time
114+
v float64
115+
}{{hour, 10}, {badHour, 7}} {
116+
points, err := s.LoadSeriesBucketsOrRaw("meter", "grid_w", want.at.UnixMilli(), want.at.UnixMilli()+HistoryHourResolutionMS-1, 0)
117+
if err != nil || len(points) != 1 || points[0].V != want.v {
118+
t.Fatalf("boot %d: hour %s = %+v, %v; want one point of %v", boot, want.at, points, err, want.v)
119+
}
120+
}
121+
from, to := good.UnixMilli(), bad.Add(48*time.Hour).UnixMilli()
122+
for _, maxPoints := range []int{0, 200} {
123+
for _, metric := range []string{"grid_w", "other_w"} {
124+
if _, err := s.LoadSeriesBucketsOrRaw("meter", metric, from, to, maxPoints); err != nil {
125+
t.Fatalf("boot %d: %s range over the damaged day, maxPoints=%d: %v", boot, metric, maxPoints, err)
126+
}
127+
}
128+
}
129+
now := time.Now()
130+
if _, err := s.LoadSeriesBuckets("meter", "grid_w", now.Add(-365*24*time.Hour).UnixMilli(), now.UnixMilli(), 365); err != nil {
131+
t.Fatalf("boot %d: year chart: %v", boot, err)
132+
}
133+
}
134+
})
135+
}
136+
}
137+
138+
// An operator who puts back a good copy of the damaged file must get a
139+
// clean boot. Had the legacy copy imported the same hour, the restored file
140+
// would conflict with it and stop boot again.
141+
func TestAbsorbColdHistoryImportsRestoredBucketFile(t *testing.T) {
142+
s, cold, day := unreadableColdFixture(t)
143+
at := day.Add(2 * time.Hour)
144+
orig := dayFile(t, cold, day, ".buckets.parquet")
145+
legacy := dayFile(t, cold, day, ".legacy-buckets.parquet")
146+
full := []metricBucket{unreadableBucket(at, 3), unreadableBucket(at.Add(time.Minute), 4)}
147+
if err := writeTestBuckets(orig, full); err != nil {
148+
t.Fatal(err)
149+
}
150+
good, err := os.ReadFile(orig)
151+
if err != nil {
152+
t.Fatal(err)
153+
}
154+
if err := os.Truncate(orig, int64(len(good)/2)); err != nil {
155+
t.Fatal(err)
156+
}
157+
if err := writeTestBuckets(legacy, full[:1]); err != nil {
158+
t.Fatal(err)
159+
}
160+
if err := s.AbsorbColdHistory(context.Background(), cold); err != nil {
161+
t.Fatalf("first boot: %v", err)
162+
}
163+
if err := os.Remove(orig + unreadableColdSuffix); err != nil {
164+
t.Fatal(err)
165+
}
166+
if err := os.WriteFile(orig, good, 0o644); err != nil {
167+
t.Fatal(err)
168+
}
169+
if err := s.AbsorbColdHistory(context.Background(), cold); err != nil {
170+
t.Fatalf("boot after restoring the file: %v", err)
171+
}
172+
assertFile(t, orig, false)
173+
assertFile(t, legacy, false)
174+
if listed := listedUnreadable(s); len(listed) != 0 {
175+
t.Fatalf("unreadable_cold_files = %v after recovery", listed)
176+
}
177+
points, err := s.LoadSeriesBucketsOrRaw("meter", "grid_w", at.UnixMilli(), at.UnixMilli()+HistoryHourResolutionMS-1, 0)
178+
if err != nil || len(points) != 1 || points[0].N != 2 {
179+
t.Fatalf("restored hour = %+v, %v; want one point from both minutes", points, err)
180+
}
181+
}

‎go/internal/state/history_plain.go‎

Lines changed: 77 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"math"
1010
"os"
1111
"path/filepath"
12+
"slices"
1213
"strconv"
1314
"strings"
1415
"time"
@@ -200,6 +201,13 @@ func (s *Store) RetireRawHistory(ctx context.Context) error {
200201
// goes with them, and so does a sample day whose hourly summary receipt
201202
// matches the file. Other sample days stay readable until the background
202203
// rollup summarizes them. Diagnostics files are left in place.
204+
//
205+
// An unreadable bucket file never stops boot. It moves aside under
206+
// unreadableColdSuffix, out of every reader's glob, so reads for that day
207+
// still work. It stays on disk, and HistoryBackend lists it. The day's legacy
208+
// copy stays on disk unimported and serves reads. A good copy of the original
209+
// restored under its normal name imports on the next boot, and the legacy
210+
// copy then goes without a conflict.
203211
func (s *Store) AbsorbColdHistory(ctx context.Context, cold string) error {
204212
if s == nil || s.history == nil || strings.TrimSpace(cold) == "" {
205213
return nil
@@ -213,19 +221,49 @@ func (s *Store) AbsorbColdHistory(ctx context.Context, cold string) error {
213221
return err
214222
}
215223
var imported int
224+
unreadable := map[string]bool{}
225+
// held lists files kept at their normal name: an unreadable file that
226+
// could not move aside, or the legacy copy of an unreadable original.
227+
held := map[string]bool{}
228+
skip := func(path string, err error) {
229+
unreadable[path] = true
230+
aside := path + unreadableColdSuffix
231+
if _, statErr := os.Lstat(aside); errors.Is(statErr, os.ErrNotExist) {
232+
renameErr := os.Rename(path, aside)
233+
if renameErr == nil {
234+
slog.Warn("history bucket file unreadable; moved aside", "path", aside, "err", err)
235+
return
236+
}
237+
slog.Warn("history bucket file could not move aside", "path", path, "err", renameErr)
238+
}
239+
slog.Warn("history bucket file unreadable; kept in place", "path", path, "err", err)
240+
held[path] = true
241+
}
216242
for _, path := range paths {
217243
if err := ctx.Err(); err != nil {
218244
return err
219245
}
220246
if strings.Contains(path, ".legacy-buckets.") {
221247
original := strings.Replace(path, ".legacy-buckets.parquet", ".buckets.parquet", 1)
222-
if _, statErr := os.Stat(original); statErr == nil {
248+
if _, statErr := os.Stat(original); statErr == nil && !unreadable[original] {
249+
continue
250+
}
251+
// Importing these hours would conflict with a good copy of the
252+
// original restored later, and stop that boot. Keep the file;
253+
// it serves reads for the day until then.
254+
if _, statErr := os.Lstat(original + unreadableColdSuffix); statErr == nil || unreadable[original] {
255+
slog.Warn("history legacy bucket file kept while its original is unreadable", "path", path)
256+
held[path] = true
223257
continue
224258
}
225259
}
226260
digest, err := historyFileHashContext(ctx, path)
227261
if err != nil {
228-
return err
262+
if ctx.Err() != nil {
263+
return ctx.Err()
264+
}
265+
skip(path, err)
266+
continue
229267
}
230268
receipt := "plain-archive:" + digest
231269
hours := map[string]*namedHour{}
@@ -240,7 +278,11 @@ func (s *Store) AbsorbColdHistory(ctx context.Context, cold string) error {
240278
h.sources = append(h.sources, b.BucketSummary)
241279
return nil
242280
}); err != nil {
243-
return fmt.Errorf("read %s: %w", filepath.Base(path), err)
281+
if ctx.Err() != nil {
282+
return ctx.Err()
283+
}
284+
skip(path, err)
285+
continue
244286
}
245287
if err := s.writeHourBuckets(ctx, hours, receipt); err != nil {
246288
return err
@@ -251,20 +293,50 @@ func (s *Store) AbsorbColdHistory(ctx context.Context, cold string) error {
251293
if err != nil {
252294
return err
253295
}
296+
sampleDays := len(keep)
297+
// Earlier boots moved unreadable files aside; list them on every boot.
298+
aside, err := filepath.Glob(filepath.Join(cold, "[0-9][0-9][0-9][0-9]", "[0-9][0-9]", "[0-9][0-9].*"+unreadableColdSuffix))
299+
if err != nil {
300+
return err
301+
}
302+
var kept []string
303+
for _, path := range aside {
304+
keep[path] = true
305+
kept = append(kept, coldRel(path))
306+
}
307+
for path := range held {
308+
keep[path] = true
309+
if unreadable[path] {
310+
kept = append(kept, coldRel(path))
311+
}
312+
}
313+
slices.Sort(kept)
314+
s.maintenanceStatusMu.Lock()
315+
s.coldUnreadable = kept
316+
s.maintenanceStatusMu.Unlock()
254317
if err := removeColdHistory(cold, keep); err != nil {
255318
return err
256319
}
257320
removed, err := removeRetiredHistory(s.historyPath)
258321
if err != nil {
259322
return err
260323
}
261-
if len(keep) > 0 {
262-
slog.Warn("history sample days kept until they have hourly summaries", "days", len(keep))
324+
if sampleDays > 0 {
325+
slog.Warn("history sample days kept until they have hourly summaries", "days", sampleDays)
263326
}
264327
slog.Info("history cold files absorbed", "days", imported, "retired_files", removed)
265328
return nil
266329
}
267330

331+
// unreadableColdSuffix marks a bucket file that failed to read. The name no
332+
// longer ends in .parquet, so no reader, importer or converter opens it.
333+
const unreadableColdSuffix = ".unreadable"
334+
335+
// coldRel names a cold file as YYYY/MM/<file>.
336+
func coldRel(path string) string {
337+
return filepath.Base(filepath.Dir(filepath.Dir(path))) + "/" + filepath.Base(filepath.Dir(path)) + "/" + filepath.Base(path)
338+
}
339+
268340
// A sample day counts as absorbed only through the summary receipt for this
269341
// exact file. 2.x wrote these days and never summarized them.
270342
func (s *Store) unsummarizedSampleDays(ctx context.Context, cold string) (map[string]bool, error) {

‎go/internal/state/history_sqlite.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import (
1313
"math"
1414
"os"
1515
"path/filepath"
16+
"slices"
1617
"strings"
1718
"time"
1819

@@ -463,6 +464,11 @@ func historyFloatBits(value float64) uint64 {
463464
func (s *Store) HistoryBackend() map[string]any {
464465
info := map[string]any{"engine": "sqlite", "archive": "sqlite", "file": filepath.Base(s.historyPath), "writer": s.HistoryWriterStatus(), "migration": s.HistoryMigrationStatus(), "series_hour": s.SeriesHourBackfillStatus()}
465466
info["maintenance"] = s.HistoryMaintenanceStatus()
467+
s.maintenanceStatusMu.Lock()
468+
if len(s.coldUnreadable) > 0 {
469+
info["unreadable_cold_files"] = slices.Clone(s.coldUnreadable)
470+
}
471+
s.maintenanceStatusMu.Unlock()
466472
if s.aggregateHistory.Load() {
467473
info["policy"] = map[string]any{"id": "sqlite-v1", "recent_resolution_ms": HistoryResolutionMS, "recent_retention_hours": int(plainTenSecondKeep.Hours()), "archive_resolution_ms": ArchiveResolutionMS, "archive_minute_days": int(plainMinuteKeep.Hours() / 24), "older_resolution_ms": HistoryHourResolutionMS, "detailed_retention_days": int(plainHourKeep.Hours() / 24)}
468474
}

‎go/internal/state/store.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,7 @@ type Store struct {
8787

8888
maintenanceStatusMu sync.Mutex
8989
maintenanceStatus HistoryMaintenanceStatus
90+
coldUnreadable []string // guarded by maintenanceStatusMu
9091
// offlineBackup is set by OpenBackupSource. Live Core backups yield
9192
// between copy batches so goal and control writes stay within latency
9293
// limits; the offline helper must not inherit that 100 ms live pause.

0 commit comments

Comments
 (0)