wal.go raw
1 package store
2
3 import (
4 "encoding/binary"
5 "errors"
6 "fmt"
7 "hash/crc32"
8 "io"
9 "os"
10 "path/filepath"
11 "sort"
12 "strings"
13 )
14
15 // The WAL is a sequence of segments named seg-NNN.dat. Each record is
16 //
17 // uint32 length | uint32 crc32(payload) | payload
18 //
19 // A serial (segment<<32 | offset) is the physical address of the record's
20 // header, so reading an event is one open segment, one ReadAt for the header
21 // and one ReadAt for the payload. The log is never rewritten.
22 const (
23 walHeaderSize = 8
24 // MaxSegSize bounds a segment so its offset fits in the 32 low bits of a
25 // serial. 256 segments of 1 GiB is the ceiling of the uint40 address.
26 MaxSegSize = 1 << 30
27 // maxRecord bounds a single record so a corrupt length cannot ask for an
28 // absurd allocation.
29 maxRecord = 64 << 20
30 syncEvery = 100
31 )
32
33 var (
34 ErrCorruptWAL = errors.New("store: corrupt WAL record")
35 ErrShortRead = errors.New("store: short WAL read")
36 )
37
38 type wal struct {
39 dir string
40 segs []*os.File
41 cur int
42 off int64
43 wrote int
44 noSync bool
45 }
46
47 func openWAL(dir string) (*wal, error) {
48 if err := os.MkdirAll(dir, 0o755); err != nil {
49 return nil, err
50 }
51 entries, err := os.ReadDir(dir)
52 if err != nil {
53 return nil, err
54 }
55 var names []string
56 for _, e := range entries {
57 n := e.Name()
58 if strings.HasPrefix(n, "seg-") && strings.HasSuffix(n, ".dat") {
59 names = append(names, n)
60 }
61 }
62 sort.Strings(names)
63 w := &wal{dir: dir}
64 for _, name := range names {
65 f, err := os.OpenFile(filepath.Join(dir, name), os.O_RDWR, 0o644)
66 if err != nil {
67 w.Close()
68 return nil, err
69 }
70 w.segs = append(w.segs, f)
71 }
72 if len(w.segs) == 0 {
73 if err := w.newSeg(); err != nil {
74 return nil, err
75 }
76 } else {
77 w.cur = len(w.segs) - 1
78 info, err := w.segs[w.cur].Stat()
79 if err != nil {
80 return nil, err
81 }
82 if err := w.repairTail(info.Size()); err != nil {
83 return nil, err
84 }
85 }
86 return w, nil
87 }
88
89 // repairTail finds the end of the last complete record in the current segment
90 // and truncates a torn tail left by a crash.
91 func (w *wal) repairTail(size int64) error {
92 f := w.segs[w.cur]
93 hdr := make([]byte, walHeaderSize)
94 payload := make([]byte, 0, 64<<10)
95 off := int64(0)
96 for off < size {
97 if size-off < walHeaderSize {
98 break
99 }
100 if _, err := f.ReadAt(hdr, off); err != nil {
101 break
102 }
103 length := int64(binary.BigEndian.Uint32(hdr[0:4]))
104 if length <= 0 || length > maxRecord || off+walHeaderSize+length > size {
105 break
106 }
107 if int64(cap(payload)) < length {
108 payload = make([]byte, length)
109 }
110 payload = payload[:length]
111 if _, err := f.ReadAt(payload, off+walHeaderSize); err != nil {
112 break
113 }
114 if crc32.ChecksumIEEE(payload) != binary.BigEndian.Uint32(hdr[4:8]) {
115 break
116 }
117 off += walHeaderSize + length
118 }
119 if off != size {
120 if err := f.Truncate(off); err != nil {
121 return err
122 }
123 if _, err := f.Seek(off, io.SeekStart); err != nil {
124 return err
125 }
126 }
127 w.off = off
128 return nil
129 }
130
131 func (w *wal) newSeg() error {
132 n := len(w.segs)
133 f, err := os.OpenFile(filepath.Join(w.dir, fmt.Sprintf("seg-%03d.dat", n)), os.O_RDWR|os.O_CREATE, 0o644)
134 if err != nil {
135 return err
136 }
137 w.segs = append(w.segs, f)
138 w.cur = n
139 w.off = 0
140 return nil
141 }
142
143 // Pos returns the serial the next record will be written at.
144 func (w *wal) Pos() uint64 { return uint64(w.cur)<<32 | uint64(w.off) }
145
146 // Append writes a record and returns its serial.
147 func (w *wal) Append(data []byte) (uint64, error) {
148 need := int64(walHeaderSize + len(data))
149 if w.off+need > MaxSegSize {
150 if err := w.Sync(); err != nil {
151 return 0, err
152 }
153 if err := w.newSeg(); err != nil {
154 return 0, err
155 }
156 }
157 ser := uint64(w.cur)<<32 | uint64(w.off)
158 hdr := make([]byte, walHeaderSize)
159 binary.BigEndian.PutUint32(hdr[0:4], uint32(len(data)))
160 binary.BigEndian.PutUint32(hdr[4:8], crc32.ChecksumIEEE(data))
161 f := w.segs[w.cur]
162 if _, err := f.WriteAt(hdr, w.off); err != nil {
163 return 0, err
164 }
165 if _, err := f.WriteAt(data, w.off+walHeaderSize); err != nil {
166 return 0, err
167 }
168 w.off += need
169 w.wrote++
170 if !w.noSync && w.wrote >= syncEvery {
171 w.wrote = 0
172 if err := f.Sync(); err != nil {
173 return 0, err
174 }
175 }
176 return ser, nil
177 }
178
179 // Read returns the payload stored at ser.
180 func (w *wal) Read(ser uint64) ([]byte, error) {
181 seg := int(ser >> 32)
182 off := int64(ser & 0xffffffff)
183 if seg < 0 || seg >= len(w.segs) {
184 return nil, ErrShortRead
185 }
186 f := w.segs[seg]
187 hdr := make([]byte, walHeaderSize)
188 if _, err := f.ReadAt(hdr, off); err != nil {
189 return nil, err
190 }
191 length := int64(binary.BigEndian.Uint32(hdr[0:4]))
192 if length <= 0 || length > maxRecord {
193 return nil, ErrCorruptWAL
194 }
195 data := make([]byte, length)
196 if _, err := f.ReadAt(data, off+walHeaderSize); err != nil {
197 return nil, err
198 }
199 if crc32.ChecksumIEEE(data) != binary.BigEndian.Uint32(hdr[4:8]) {
200 return nil, ErrCorruptWAL
201 }
202 return data, nil
203 }
204
205 // ForEachFrom calls fn for every record at or after pos. fn returning false
206 // stops the scan.
207 func (w *wal) ForEachFrom(pos uint64, fn func(ser uint64, data []byte) bool) error {
208 seg := int(pos >> 32)
209 off := int64(pos & 0xffffffff)
210 if seg >= len(w.segs) {
211 return nil
212 }
213 hdr := make([]byte, walHeaderSize)
214 var payload []byte
215 for ; seg < len(w.segs); seg++ {
216 f := w.segs[seg]
217 if seg == int(pos>>32) {
218 off = int64(pos & 0xffffffff)
219 } else {
220 off = 0
221 }
222 info, err := f.Stat()
223 if err != nil {
224 return err
225 }
226 size := info.Size()
227 for off < size {
228 if size-off < walHeaderSize {
229 break
230 }
231 if _, err := f.ReadAt(hdr, off); err != nil {
232 return err
233 }
234 length := int64(binary.BigEndian.Uint32(hdr[0:4]))
235 if length <= 0 || length > maxRecord || off+walHeaderSize+length > size {
236 break
237 }
238 if int64(cap(payload)) < length {
239 payload = make([]byte, length)
240 }
241 payload = payload[:length]
242 if _, err := f.ReadAt(payload, off+walHeaderSize); err != nil {
243 return err
244 }
245 if crc32.ChecksumIEEE(payload) != binary.BigEndian.Uint32(hdr[4:8]) {
246 break
247 }
248 ser := uint64(seg)<<32 | uint64(off)
249 if !fn(ser, payload) {
250 return nil
251 }
252 off += walHeaderSize + length
253 }
254 }
255 return nil
256 }
257
258 func (w *wal) Sync() error {
259 if len(w.segs) == 0 {
260 return nil
261 }
262 return w.segs[w.cur].Sync()
263 }
264
265 func (w *wal) Close() error {
266 var first error
267 for _, f := range w.segs {
268 if err := f.Sync(); err != nil && first == nil {
269 first = err
270 }
271 if err := f.Close(); err != nil && first == nil {
272 first = err
273 }
274 }
275 w.segs = nil
276 return first
277 }
278