-
Notifications
You must be signed in to change notification settings - Fork 394
/
v1.go
276 lines (237 loc) · 7.86 KB
/
v1.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
// Copyright (C) 2020 Storj Labs, Inc.
// See LICENSE for copying information.
package ordersfile
import (
"bufio"
"bytes"
"encoding/binary"
"errors"
"hash/crc32"
"io"
"os"
"time"
"github.com/zeebo/errs"
"storj.io/common/memory"
"storj.io/common/pb"
"storj.io/common/storj"
"storj.io/storj/private/date"
)
var (
// fileMagic used to identify header of file.
// "0ddba11 acc01ade5".
fileMagic = [8]byte{0x0d, 0xdb, 0xa1, 0x1a, 0xcc, 0x01, 0xad, 0xe5}
// entryHeader is 8 bytes that appears before every order/limit in a V1 file.
// "5ca1ab1e ba5eba11".
entryHeader = [8]byte{0x5c, 0xa1, 0xab, 0x1e, 0xba, 0x5e, 0xba, 0x11}
// entryFooter is 8 bytes that appears after every order/limit in a V1 file.
// "feed 1 f00d 1 c0ffee".
entryFooter = [8]byte{0xfe, 0xed, 0x1f, 0x00, 0xd1, 0xc0, 0xff, 0xee}
)
// fileV1 is a version 1 orders file.
type fileV1 struct {
f *os.File
br *bufio.Reader
}
// OpenWritableV1 opens for writing the unsent or archived orders file at a given path.
// If the file is new, the file header is written.
func OpenWritableV1(path string, satelliteID storj.NodeID, creationTime time.Time) (Writable, error) {
// create file if not exists or append
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return nil, Error.Wrap(err)
}
of := &fileV1{
f: f,
}
currentPos, err := of.f.Seek(0, io.SeekCurrent)
if err != nil {
return nil, Error.Wrap(err)
}
if currentPos == 0 {
err = of.writeHeader(satelliteID, creationTime)
if err != nil {
return of, err
}
}
return of, nil
}
// writeHeader writes file header as [filemagic][satellite ID][creation hour].
func (of *fileV1) writeHeader(satelliteID storj.NodeID, creationTime time.Time) error {
toWrite := fileMagic[:]
toWrite = append(toWrite, satelliteID.Bytes()...)
creationHour := date.TruncateToHourInNano(creationTime)
creationHourBytes := [8]byte{}
binary.LittleEndian.PutUint64(creationHourBytes[:], uint64(creationHour))
toWrite = append(toWrite, creationHourBytes[:]...)
if _, err := of.f.Write(toWrite); err != nil {
return Error.New("Couldn't write file header: %w", err)
}
return nil
}
// OpenReadableV1 opens for reading the unsent or archived orders file at a given path.
func OpenReadableV1(path string) (Readable, error) {
f, err := os.Open(path)
if err != nil {
return nil, Error.Wrap(err)
}
return &fileV1{
f: f,
// buffered reader is used to search for entryHeader that precedes limit and order.
br: bufio.NewReader(f),
}, nil
}
// Append writes limit and order to the file as
// [entryHeader][limitSize][limitBytes][orderSize][orderBytes][checksum][entryFooter].
func (of *fileV1) Append(info *Info) error {
toWrite := entryHeader[:]
limitSerialized, err := pb.Marshal(info.Limit)
if err != nil {
return Error.Wrap(err)
}
orderSerialized, err := pb.Marshal(info.Order)
if err != nil {
return Error.Wrap(err)
}
limitSizeBytes := [2]byte{}
binary.LittleEndian.PutUint16(limitSizeBytes[:], uint16(len(limitSerialized)))
orderSizeBytes := [2]byte{}
binary.LittleEndian.PutUint16(orderSizeBytes[:], uint16(len(orderSerialized)))
toWrite = append(toWrite, limitSizeBytes[:]...)
toWrite = append(toWrite, limitSerialized...)
toWrite = append(toWrite, orderSizeBytes[:]...)
toWrite = append(toWrite, orderSerialized...)
checksumInt := crc32.ChecksumIEEE(toWrite[len(entryHeader):])
checksumBytes := [4]byte{}
binary.LittleEndian.PutUint32(checksumBytes[:], checksumInt)
toWrite = append(toWrite, checksumBytes[:]...)
toWrite = append(toWrite, entryFooter[:]...)
if _, err = of.f.Write(toWrite); err != nil {
return Error.New("Couldn't write serialized order and limit: %w", err)
}
return nil
}
// ReadOne reads one entry from the file.
// It returns ErrEntryCorrupt upon finding a corrupt limit/order combo. On next call after a corrupt entry, it will find the next valid order.
func (of *fileV1) ReadOne() (info *Info, err error) {
// attempt to read an order/limit; if corrupted, keep trying until EOF or uncorrupted pair found.
// start position will be the position of the of.f cursor minus the number of unread buffered bytes in of.br
startPosition, err := of.f.Seek(0, io.SeekCurrent)
if err != nil {
return nil, Error.Wrap(err)
}
startPosition -= int64(of.br.Buffered())
defer func() {
// Treat all non-EOF errors as corrupt entry errors so that ReadOne is called again.
if err != nil && !errors.Is(err, io.EOF) {
// seek forward by len(entryHeader) bytes for next iteration.
nextStartPosition := startPosition + int64(len(entryHeader))
_, seekErr := of.f.Seek(nextStartPosition, io.SeekStart)
if seekErr != nil {
err = errs.Combine(err, seekErr)
}
of.br.Reset(of.f)
err = ErrEntryCorrupt.Wrap(err)
}
}()
err = of.gotoNextEntry()
if err != nil {
return nil, err
}
limitSizeBytes := [2]byte{}
_, err = io.ReadFull(of.br, limitSizeBytes[:])
if err != nil {
return nil, Error.Wrap(err)
}
limitSize := binary.LittleEndian.Uint16(limitSizeBytes[:])
if limitSize > uint16(orderLimitSizeCap) {
return nil, ErrEntryCorrupt.New("invalid limit size: %d is over the maximum %d", limitSize, orderLimitSizeCap)
}
limitSerialized := make([]byte, limitSize)
_, err = io.ReadFull(of.br, limitSerialized)
if err != nil {
return nil, Error.Wrap(err)
}
orderSizeBytes := [2]byte{}
_, err = io.ReadFull(of.br, orderSizeBytes[:])
if err != nil {
return nil, Error.Wrap(err)
}
orderSize := binary.LittleEndian.Uint16(orderSizeBytes[:])
if orderSize > uint16(orderSizeCap) {
return nil, ErrEntryCorrupt.New("invalid order size: %d is over the maximum %d", orderSize, orderSizeCap)
}
orderSerialized := make([]byte, orderSize)
_, err = io.ReadFull(of.br, orderSerialized)
if err != nil {
return nil, Error.Wrap(err)
}
// read checksum
checksumBytes := [4]byte{}
_, err = io.ReadFull(of.br, checksumBytes[:])
if err != nil {
return nil, Error.Wrap(err)
}
expectedChecksum := binary.LittleEndian.Uint32(checksumBytes[:])
actualChecksum := uint32(0)
actualChecksum = crc32.Update(actualChecksum, crc32.IEEETable, limitSizeBytes[:])
actualChecksum = crc32.Update(actualChecksum, crc32.IEEETable, limitSerialized)
actualChecksum = crc32.Update(actualChecksum, crc32.IEEETable, orderSizeBytes[:])
actualChecksum = crc32.Update(actualChecksum, crc32.IEEETable, orderSerialized)
if expectedChecksum != actualChecksum {
return nil, Error.New("checksum does not match")
}
footerBytes := [len(entryFooter)]byte{}
_, err = io.ReadFull(of.br, footerBytes[:])
if err != nil {
return nil, Error.Wrap(err)
}
if !bytes.Equal(entryFooter[:], footerBytes[:]) {
return nil, Error.New("footer bytes do not match")
}
limit := &pb.OrderLimit{}
err = pb.Unmarshal(limitSerialized, limit)
if err != nil {
return nil, Error.Wrap(err)
}
order := &pb.Order{}
err = pb.Unmarshal(orderSerialized, order)
if err != nil {
return nil, Error.Wrap(err)
}
return &Info{
Limit: limit,
Order: order,
}, nil
}
func (of *fileV1) gotoNextEntry() error {
// search file for next occurrence of entry header, or until EOF
for {
searchBufSize := 2 * memory.KiB.Int()
nextBufferBytes, err := of.br.Peek(searchBufSize)
// if the buffered reader hits an EOF, the buffered data may still
// contain a full entry, so do not return unless there is definitely no entry
if errors.Is(err, io.EOF) && len(nextBufferBytes) <= len(entryHeader) {
return err
} else if err != nil && !errors.Is(err, io.EOF) {
return Error.Wrap(err)
}
i := bytes.Index(nextBufferBytes, entryHeader[:])
if i > -1 {
_, err = of.br.Discard(i + len(entryHeader))
if err != nil {
return Error.Wrap(err)
}
break
}
// entry header not found; discard all but last (len(entryHeader)-1) bytes for next iteration
_, err = of.br.Discard(len(nextBufferBytes) - len(entryHeader) + 1)
if err != nil {
return Error.Wrap(err)
}
}
return nil
}
// Close closes the file.
func (of *fileV1) Close() error {
return of.f.Close()
}