"strings"
)
-// position names one line of a day file: the day as YYYYMMDD, the byte offset
+// Position names one line of a day file: the day as YYYYMMDD, the byte offset
// of the line, and its length with the newline.
-type position struct {
+type Position struct {
day uint32
offset uint64
length uint32
}
-// journal appends JSON lines to one file for each day in dir, named
+// Journal appends JSON lines to one file for each day in dir, named
// YYYY-MM-DD.jsonl, and reads them back by their position. It keeps the file of
// the last day open for writes and each file that it read open for reads.
-type journal struct {
+type Journal struct {
dir string
file *os.File
day uint32
// Write appends line to the file of day, YYYY-MM-DD, with one write to the
// operating system, without a buffer. A crash of the process then loses
// nothing. It returns the position of the line.
-func (j *journal) Write(day string, line []byte) (position, error) {
- number, err := dayNumber(day)
+func (j *Journal) Write(day string, line []byte) (Position, error) {
+ number, err := journalDayNumber(day)
if err != nil {
- return position{}, err
+ return Position{}, err
}
if number != j.day {
err = j.closeWriter()
if err != nil {
- return position{}, err
+ return Position{}, err
}
j.file, err = os.OpenFile(filepath.Join(j.dir, day+".jsonl"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
if err != nil {
- return position{}, fmt.Errorf("open day file: %w", err)
+ return Position{}, fmt.Errorf("open day file: %w", err)
}
info, err := j.file.Stat()
if err != nil {
- return position{}, fmt.Errorf("read day file size: %w", err)
+ return Position{}, fmt.Errorf("read day file size: %w", err)
}
j.day = number
_, err = j.file.Write(line)
if err != nil {
- return position{}, fmt.Errorf("write line: %w", err)
+ return Position{}, fmt.Errorf("write line: %w", err)
}
- place := position{
+ place := Position{
day: number,
offset: j.size,
length: uint32(len(line)),
return place, nil
}
-func (j *journal) closeWriter() error {
- if j.file == nil {
- return nil
- }
-
- err := j.file.Close()
- j.file = nil
- j.day = 0
- if err != nil {
- return fmt.Errorf("close day file: %w", err)
- }
-
- return nil
-}
-
// Read returns the line at place without its newline.
-func (j *journal) Read(place position) ([]byte, error) {
+func (j *Journal) Read(place Position) ([]byte, error) {
reader, open := j.readers[place.day]
if !open {
var err error
- reader, err = os.Open(filepath.Join(j.dir, dayName(place.day)+".jsonl"))
+
+ reader, err = os.Open(filepath.Join(j.dir, journalDayName(place.day)+".jsonl"))
if err != nil {
return nil, fmt.Errorf("open day file: %w", err)
}
line := make([]byte, place.length)
_, err := reader.ReadAt(line, int64(place.offset))
if err != nil {
- return nil, fmt.Errorf("read line of %s: %w", dayName(place.day), err)
+ return nil, fmt.Errorf("read line of %s: %w", journalDayName(place.day), err)
}
if len(line) == 0 || line[len(line)-1] != '\n' {
- return nil, fmt.Errorf("line of %s at %d does not end its record", dayName(place.day), place.offset)
+ return nil, fmt.Errorf("line of %s at %d does not end its record", journalDayName(place.day), place.offset)
}
return line[:len(line)-1], nil
}
// Close closes the open day files.
-func (j *journal) Close() error {
+func (j *Journal) Close() error {
err := j.closeWriter()
for day, reader := range j.readers {
err = errors.Join(err, reader.Close())
+
delete(j.readers, day)
}
}
// Scan gives each line of the day files from place on to add, with its
-// position, oldest day first. The zero position starts at the first line.
-func (j *journal) Scan(ctx context.Context, place position, add func([]byte, position) error) error {
+// position, oldest day first. The zero position starts at the first line. A
+// crash during a write can leave a part of a line at the end of the last day
+// file. Its write never returned, so Scan cuts it off.
+func (j *Journal) Scan(ctx context.Context, place Position, add func([]byte, Position) error) error {
paths, err := filepath.Glob(filepath.Join(j.dir, "*.jsonl"))
if err != nil {
return fmt.Errorf("list day files: %w", err)
}
slices.Sort(paths)
- for _, path := range paths {
- day, err := dayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl"))
+
+ for n, path := range paths {
+ day, err := journalDayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl"))
if err != nil {
return err
}
offset = place.offset
}
- err = scanFile(ctx, path, day, offset, add)
+ err = journalScanFile(ctx, path, day, offset, n == len(paths)-1, add)
if err != nil {
return err
}
return nil
}
-// scanFile gives each line of the day file at path from offset on to add.
-func scanFile(ctx context.Context, path string, day uint32, offset uint64, add func([]byte, position) error) (err error) {
+func (j *Journal) closeWriter() error {
+ if j.file == nil {
+ return nil
+ }
+
+ err := j.file.Close()
+ j.file = nil
+ j.day = 0
+ if err != nil {
+ return fmt.Errorf("close day file: %w", err)
+ }
+
+ return nil
+}
+
+// journalDayNumber returns the day YYYY-MM-DD as the number YYYYMMDD.
+func journalDayNumber(day string) (uint32, error) {
+ number, err := strconv.ParseUint(strings.ReplaceAll(day, "-", ""), 10, 32)
+ if err != nil || len(day) != len("2006-01-02") {
+ return 0, fmt.Errorf("invalid day %q", day)
+ }
+
+ return uint32(number), nil
+}
+
+// journalDayName returns the day number YYYYMMDD as YYYY-MM-DD.
+func journalDayName(day uint32) string {
+ return fmt.Sprintf("%04d-%02d-%02d", day/10000, day/100%100, day%100)
+}
+
+// journalScanFile gives each line of the day file at path from offset on to
+// add. In the last day file, it cuts a part of a line at the end.
+func journalScanFile(ctx context.Context, path string, day uint32, offset uint64, last bool, add func([]byte, Position) error) (err error) {
file, err := os.Open(path)
if err != nil {
return fmt.Errorf("open day file: %w", err)
return nil
}
+ if errors.Is(err, io.EOF) && last {
+ return journalCutFile(path, offset)
+ }
+
if errors.Is(err, io.EOF) {
return fmt.Errorf("%s ends in an incomplete line at %d", filepath.Base(path), offset)
}
return fmt.Errorf("read %s: %w", filepath.Base(path), err)
}
- place := position{
+ place := Position{
day: day,
offset: offset,
length: uint32(len(line)),
}
}
-// dayNumber returns the day YYYY-MM-DD as the number YYYYMMDD.
-func dayNumber(day string) (uint32, error) {
- number, err := strconv.ParseUint(strings.ReplaceAll(day, "-", ""), 10, 32)
- if err != nil || len(day) != len("2006-01-02") {
- return 0, fmt.Errorf("invalid day %q", day)
+// journalCutFile cuts the file at path to size bytes.
+func journalCutFile(path string, size uint64) error {
+ err := os.Truncate(path, int64(size))
+ if err != nil {
+ return fmt.Errorf("cut incomplete line: %w", err)
}
- return uint32(number), nil
-}
-
-// dayName returns the day number YYYYMMDD as YYYY-MM-DD.
-func dayName(day uint32) string {
- return fmt.Sprintf("%04d-%02d-%02d", day/10000, day/100%100, day%100)
+ return nil
}
projects / onechat / blobdiff