(ctx context.Context)
| 83 | } |
| 84 | |
| 85 | func (s *Store) load(ctx context.Context) error { |
| 86 | head, err := s.journal.ReadHead(ctx) |
| 87 | if err != nil { |
| 88 | return err |
| 89 | } |
| 90 | s.mu.RLock() |
| 91 | current := s.at |
| 92 | s.mu.RUnlock() |
| 93 | if head == current { |
| 94 | return nil |
| 95 | } |
| 96 | unmarshaler := s.newUnmarshaler() |
| 97 | at, table, err := s.getSnapshot(ctx, unmarshaler) |
| 98 | if err != nil && !errors.Is(err, fs.ErrNotExist) { |
| 99 | s.logger.Error("Loading snapshot", zap.Error(err)) |
| 100 | } |
| 101 | if at == Nil { |
| 102 | // Load base if it exists. |
| 103 | tail, base, err := s.journal.ReadTail(ctx) |
| 104 | if err != nil { |
| 105 | return err |
| 106 | } |
| 107 | if table, err = s.loadBase(ctx, base, unmarshaler); err != nil { |
| 108 | return err |
| 109 | } |
| 110 | at = tail |
| 111 | } |
| 112 | r, err := s.journal.OpenAsBSUP(ctx, super.NewContext(), head, at) |
| 113 | if err != nil { |
| 114 | return err |
| 115 | } |
| 116 | defer r.Close() |
| 117 | for { |
| 118 | val, err := r.Read() |
| 119 | if err != nil { |
| 120 | return err |
| 121 | } |
| 122 | if val == nil { |
| 123 | now := time.Now() |
| 124 | s.mu.Lock() |
| 125 | s.table = table |
| 126 | s.at = head |
| 127 | s.loadTime = now |
| 128 | s.mu.Unlock() |
| 129 | // Reduce the amount of times we write snapshots to disk by only writing when there are |
| 130 | // more than 10 new entries since the last snapshot. |
| 131 | if head-at > 10 { |
| 132 | if err := s.putSnapshot(ctx, head, table); err != nil { |
| 133 | s.logger.Error("Storing snapshot", zap.Error(err)) |
| 134 | } |
| 135 | } |
| 136 | return nil |
| 137 | } |
| 138 | var e Entry |
| 139 | if err := unmarshaler.Unmarshal(*val, &e); err != nil { |
| 140 | return err |
| 141 | } |
| 142 | updateTable(table, e) |
no test coverage detected