X-Git-Url: http://git.giacomolli.uk/onechat/blobdiff_plain/4e3341f81d4c9b296c8fd016f4a086752d561efd..7b6d0a81e0f3c8e33b1b76d5ecbc5bdb25a87249:/log.go?ds=sidebyside diff --git a/log.go b/log.go index f1727d5..c3d7190 100644 --- a/log.go +++ b/log.go @@ -7,14 +7,10 @@ import ( "fmt" "os" "path/filepath" - "syscall" "time" "unicode/utf8" ) -// errLocked tells that another process owns the chat. -var errLocked = errors.New("another process owns the chat") - // kinds holds the kinds that a message can have. var kinds = map[string]bool{ "user": true, @@ -43,20 +39,27 @@ type Message struct { // exclusive lock on the lock file makes sure that only one process writes the // log. After a failed write, the end of the file is not known, so the log // refuses all writes until it is opened again. The caller serializes all calls. +// +// The log keeps no message in memory. The index index/main.idx holds the +// position of message i in record i, so a read of a message is one read of the +// index and one of its day file. At open the log indexes the lines after the +// last indexed one, so a missing index is built again from the day files. type Log struct { - clock func() time.Time - lock *os.File - journal journal - messages []Message - failure error + clock func() time.Time + lock *os.File + journal journal + index *records + failure error } -// openLog locks the chat in root and loads all of its messages. The clock gives -// the date of each new message. +// openLog locks the chat in root and indexes its new lines. The clock gives the +// date of each new message. func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, error) { - err := os.MkdirAll(filepath.Join(root, "main"), 0o700) - if err != nil { - return nil, fmt.Errorf("create message directory: %w", err) + for _, dir := range []string{"main", "index"} { + err := os.MkdirAll(filepath.Join(root, dir), 0o700) + if err != nil { + return nil, fmt.Errorf("create %s directory: %w", dir, err) + } } lock, err := lockFile(filepath.Join(root, "lock")) @@ -72,34 +75,70 @@ func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, er }, } - err = l.load(ctx) + l.index, err = openRecords(filepath.Join(root, "index", "main.idx"), positionBytes) + if err != nil { + return nil, errors.Join(err, l.Close()) + } + + err = l.catchUp(ctx) if err != nil { - return nil, errors.Join(fmt.Errorf("load message log: %w", err), l.Close()) + return nil, errors.Join(fmt.Errorf("index message log: %w", err), l.Close()) } return l, nil } -func (l *Log) load(ctx context.Context) error { - return readLines(ctx, l.journal.dir, func(message Message) error { - if message.ID != len(l.messages) { - return fmt.Errorf("expected message %d, found %d", len(l.messages), message.ID) +// catchUp indexes the lines after the last indexed line. +func (l *Log) catchUp(ctx context.Context) error { + start := position{} + if l.index.count > 0 { + record, _, err := l.index.Read(l.index.count - 1) + if err != nil { + return err + } + + last := decodePosition(record) + start = position{ + day: last.day, + offset: last.offset + uint64(last.length), } + } - err := validateMessage(message) + return l.journal.Scan(ctx, start, func(line []byte, place position) error { + var message Message + err := json.Unmarshal(line, &message) if err != nil { - return fmt.Errorf("message %d: %w", message.ID, err) + return err } - l.messages = append(l.messages, message) + if message.ID != l.index.count { + return fmt.Errorf("expected message %d, found %d", l.index.count, message.ID) + } - return nil + err = validateMessage(message) + if err != nil { + return fmt.Errorf("message %d: %w", message.ID, err) + } + + return l.addIndex(place) }) } +func (l *Log) addIndex(place position) error { + record := make([]byte, positionBytes) + encodePosition(record, place) + + return l.index.Append(record) +} + // Close releases the lock. The log refuses all calls after Close. func (l *Log) Close() error { err := l.journal.Close() + if l.index != nil { + err = errors.Join(err, l.index.Close()) + l.index = nil + } + if l.lock != nil { err = errors.Join(err, l.lock.Close()) l.lock = nil @@ -121,7 +160,7 @@ func (l *Log) Append(kind, text string) (Message, error) { now := l.clock() message := Message{ - ID: len(l.messages), + ID: l.index.count, Kind: kind, Text: text, Size: len(text), @@ -138,25 +177,51 @@ func (l *Log) Append(kind, text string) (Message, error) { return Message{}, fmt.Errorf("encode message: %w", err) } - err = l.journal.Write(now.Format(time.DateOnly), append(line, '\n')) + place, err := l.journal.Write(now.Format(time.DateOnly), append(line, '\n')) + if err == nil { + err = l.addIndex(place) + } + if err != nil { l.failure = err return Message{}, fmt.Errorf("write message %d: %w", message.ID, err) } - l.messages = append(l.messages, message) - return message, nil } // Len returns the number of messages in the log. func (l *Log) Len() int { - return len(l.messages) + return l.index.count } -// Message returns the message with the given ID. -func (l *Log) Message(id int) Message { - return l.messages[id] +// Message reads the message with the given ID from its day file. +func (l *Log) Message(id int) (Message, error) { + record, found, err := l.index.Read(id) + if err != nil { + return Message{}, err + } + + if !found { + return Message{}, fmt.Errorf("no message %d", id) + } + + line, err := l.journal.Read(decodePosition(record)) + if err != nil { + return Message{}, fmt.Errorf("read message %d: %w", id, err) + } + + var message Message + err = json.Unmarshal(line, &message) + if err != nil { + return Message{}, fmt.Errorf("decode message %d: %w", id, err) + } + + if message.ID != id { + return Message{}, fmt.Errorf("index of message %d points to message %d", id, message.ID) + } + + return message, nil } // validateMessage makes sure that a message has a known kind, valid UTF-8 text, @@ -181,23 +246,3 @@ func validateMessage(message Message) error { return nil } - -// lockFile opens the file at path and takes an exclusive lock on it. The lock -// stays until the file closes or the process ends. -func lockFile(path string) (*os.File, error) { - file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) - if err != nil { - return nil, fmt.Errorf("open lock file: %w", err) - } - - err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) - if errors.Is(err, syscall.EWOULDBLOCK) { - err = errLocked - } - - if err != nil { - return nil, errors.Join(fmt.Errorf("lock chat: %w", err), file.Close()) - } - - return file, nil -}