Give the rebuild its own call limit

[onechat] / journal.go

diff --git a/journal.go b/journal.go
index 3c2a7ef52e4e24cbd476f0705e5771739312c470..bb1897c68b22f8dce8acc51d34575d6cdc5b6eeb 100644 (file)
--- a/journal.go
+++ b/journal.go
@@ -3,59 +3,166 @@ package main
 import (
        "bufio"
        "context"
 import (
        "bufio"
        "context"

-       "encoding/json"

        "errors"
        "fmt"
        "errors"
        "fmt"

+       "io"

        "os"
        "path/filepath"
        "os"
        "path/filepath"

+       "slices"
+       "strconv"
+       "strings"

 )
 
 )
 

-// journal appends JSON lines to one file for each day in dir, named
-// YYYY-MM-DD.jsonl. It keeps the file of the last day open.
-type journal struct {
-       dir  string
-       file *os.File
-       day  string
+// 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 {
+       day    uint32
+       offset uint64
+       length uint32

 }
 
 }
 

-// Write appends line to the file of day and flushes it to the disk.
-func (j *journal) Write(day string, line []byte) error {
-       if day != j.day {
-               err := j.Close()
+// 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 {
+       dir     string
+       file    *os.File
+       day     uint32
+       size    uint64
+       readers map[uint32]*os.File
+}
+
+// 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 := journalDayNumber(day)
+       if err != nil {
+               return Position{}, err
+       }
+
+       if number != j.day {
+               err = j.closeWriter()

                if err != nil {
                if err != nil {

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

                }
 
                }
 

-               j.day = day
+               j.day = number
+               j.size = uint64(info.Size())

        }
 
        }
 

-       _, err := j.file.Write(line)
+       _, err = j.file.Write(line)

        if err != nil {
        if err != nil {

-               return fmt.Errorf("write line: %w", err)
+               return Position{}, fmt.Errorf("write line: %w", err)

        }
 
        }
 

-       err = j.file.Sync()
+       place := Position{
+               day:    number,
+               offset: j.size,
+               length: uint32(len(line)),
+       }
+       j.size += uint64(len(line))
+
+       return place, nil
+}
+
+// Read returns the line at place without its newline.
+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, journalDayName(place.day)+".jsonl"))
+               if err != nil {
+                       return nil, fmt.Errorf("open day file: %w", err)
+               }
+
+               if j.readers == nil {
+                       j.readers = make(map[uint32]*os.File)
+               }
+
+               j.readers[place.day] = reader
+       }
+
+       line := make([]byte, place.length)
+       _, err := reader.ReadAt(line, int64(place.offset))

        if err != nil {
        if err != nil {

-               return fmt.Errorf("flush line: %w", 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", journalDayName(place.day), place.offset)
+       }
+
+       return line[:len(line)-1], nil
+}
+
+// Close closes the open day files.
+func (j *Journal) Close() error {
+       err := j.closeWriter()
+       for day, reader := range j.readers {
+               err = errors.Join(err, reader.Close())
+
+               delete(j.readers, day)
+       }
+
+       return err
+}
+
+// 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. 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 n, path := range paths {
+               day, err := journalDayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl"))
+               if err != nil {
+                       return err
+               }
+
+               if day < place.day {
+                       continue
+               }
+
+               offset := uint64(0)
+               if day == place.day {
+                       offset = place.offset
+               }
+
+               err = journalScanFile(ctx, path, day, offset, n == len(paths)-1, add)
+               if err != nil {
+                       return err
+               }

        }
 
        return nil
 }
 
        }
 
        return nil
 }
 

-// Close closes the open day file.
-func (j *journal) Close() error {
+func (j *Journal) closeWriter() error {

        if j.file == nil {
                return nil
        }
 
        err := j.file.Close()
        j.file = nil
        if j.file == nil {
                return nil
        }
 
        err := j.file.Close()
        j.file = nil

-       j.day = ""
+       j.day = 0

        if err != nil {
                return fmt.Errorf("close day file: %w", err)
        }
        if err != nil {
                return fmt.Errorf("close day file: %w", err)
        }


@@ -63,25 +170,24 @@ func (j *journal) Close() error {
        return nil
 }
 
        return nil
 }
 

-// readLines decodes each line of the day files in dir, oldest day first, and
-// gives it to add in file order.
-func readLines[T any](ctx context.Context, dir string, add func(T) error) error {
-       paths, err := filepath.Glob(filepath.Join(dir, "*.jsonl"))
-       if err != nil {
-               return fmt.Errorf("list day files: %w", err)
+// 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)

        }
 
        }
 

-       for _, path := range paths {
-               err = readFile(ctx, path, add)
-               if err != nil {
-                       return err
-               }
-       }
+       return uint32(number), nil
+}

 
 

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

 }
 
 }
 

-func readFile[T any](ctx context.Context, path string, add func(T) error) (err error) {
+// 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)
        file, err := os.Open(path)
        if err != nil {
                return fmt.Errorf("open day file: %w", err)


@@ -91,27 +197,54 @@ func readFile[T any](ctx context.Context, path string, add func(T) error) (err e
                err = errors.Join(err, file.Close())
        }()
 
                err = errors.Join(err, file.Close())
        }()
 

-       scanner := bufio.NewScanner(file)
-       scanner.Buffer(nil, 1<<30)
-       for number := 1; scanner.Scan(); number++ {
+       _, err = file.Seek(int64(offset), io.SeekStart)
+       if err != nil {
+               return fmt.Errorf("seek day file: %w", err)
+       }
+
+       reader := bufio.NewReader(file)
+       for {

                if ctx.Err() != nil {
                        return ctx.Err()
                }
 
                if ctx.Err() != nil {
                        return ctx.Err()
                }
 

-               var value T
-               err = json.Unmarshal(scanner.Bytes(), &value)
-               if err == nil {
-                       err = add(value)
+               line, err := reader.ReadBytes('\n')
+               if errors.Is(err, io.EOF) && len(line) == 0 {
+                       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)
+               }
+
+               if err != nil {
+                       return fmt.Errorf("read %s: %w", filepath.Base(path), err)
+               }
+
+               place := Position{
+                       day:    day,
+                       offset: offset,
+                       length: uint32(len(line)),
+               }
+
+               err = add(line[:len(line)-1], place)

                if err != nil {
                if err != nil {

-                       return fmt.Errorf("%s line %d: %w", filepath.Base(path), number, err)
+                       return fmt.Errorf("%s at %d: %w", filepath.Base(path), offset, err)

                }
                }

+
+               offset += uint64(len(line))

        }
        }

+}

 
 

-       err = scanner.Err()
+// 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 {
        if err != nil {

-               return fmt.Errorf("read %s: %w", filepath.Base(path), err)
+               return fmt.Errorf("cut incomplete line: %w", err)

        }
 
        return nil
        }
 
        return nil