1package main

2

3import (

4 "bytes"

5 "context"

6 "errors"

7 "fmt"

8 "io"

9 "io/fs"

10 "log/slog"

11 "os"

12 "path/filepath"

13 "runtime"

14 "strings"

15 "syscall"

16 "time"

17

18 "github.com/mjl-/bstore"

19

20 "github.com/mjl-/mox/dmarcdb"

21 "github.com/mjl-/mox/mox-"

22 "github.com/mjl-/mox/moxvar"

23 "github.com/mjl-/mox/mtastsdb"

24 "github.com/mjl-/mox/queue"

25 "github.com/mjl-/mox/store"

26 "github.com/mjl-/mox/tlsrptdb"

27)

28

29func backupctl(ctx context.Context, ctl *ctl) {

30 /* protocol:

31 > "backup"

32 > destdir

33 > "verbose" or ""

34 < stream

35 < "ok" or error

36 */

37

38 // Convention in this function: variables containing "src" or "dst" are file system

39 // paths that can be passed to os.Open and such. Variables with dirs/paths without

40 // "src" or "dst" are incomplete paths relative to the source or destination data

41 // directories.

42

43 dstDataDir := ctl.xread()

44 verbose := ctl.xread() == "verbose"

45

46 // Set when an error is encountered. At the end, we warn if set.

47 var incomplete bool

48

49 // We'll be writing output, and logging both to mox and the ctl stream.

50 writer := ctl.writer()

51

52 // Format easily readable output for the user.

53 formatLog := func(prefix, text string, err error, attrs ...slog.Attr) []byte {

54 var b bytes.Buffer

55 fmt.Fprint(&b, prefix)

56 fmt.Fprint(&b, text)

57 if err != nil {

58 fmt.Fprint(&b, ": "+err.Error())

59 }

60 for _, a := range attrs {

61 fmt.Fprintf(&b, "; %s=%v", a.Key, a.Value)

62 }

63 fmt.Fprint(&b, "\n")

64 return b.Bytes()

65 }

66

67 // Log an error to both the mox service as the user running "mox backup".

68 pkglogx := func(prefix, text string, err error, attrs ...slog.Attr) {

69 ctl.log.Errorx(text, err, attrs...)

70

71 _, werr := writer.Write(formatLog(prefix, text, err, attrs...))

72 ctl.xcheck(werr, "write to ctl")

73 }

74

75 // Log an error but don't mark backup as failed.

76 xwarnx := func(text string, err error, attrs ...slog.Attr) {

77 pkglogx("warning: ", text, err, attrs...)

78 }

79

80 // Log an error that causes the backup to be marked as failed. We typically

81 // continue processing though.

82 xerrx := func(text string, err error, attrs ...slog.Attr) {

83 incomplete = true

84 pkglogx("error: ", text, err, attrs...)

85 }

86

87 // If verbose is enabled, log to the cli command. Always log as info level.

88 xvlog := func(text string, attrs ...slog.Attr) {

89 ctl.log.Info(text, attrs...)

90 if verbose {

91 _, werr := writer.Write(formatLog("", text, nil, attrs...))

92 ctl.xcheck(werr, "write to ctl")

93 }

94 }

95

96 if _, err := os.Stat(dstDataDir); err == nil {

97 xwarnx("destination data directory already exists", nil, slog.String("dir", dstDataDir))

98 }

99

100 srcDataDir := filepath.Clean(mox.DataDirPath("."))

101

102 // When creating a file in the destination, we first ensure its directory exists.

103 // We track which directories we created, to prevent needless syscalls.

104 createdDirs := map[string]struct{}{}

105 ensureDestDir := func(dstpath string) {

106 dstdir := filepath.Dir(dstpath)

107 if _, ok := createdDirs[dstdir]; !ok {

108 err := os.MkdirAll(dstdir, 0770)

109 if err != nil {

110 xerrx("creating directory", err)

111 }

112 createdDirs[dstdir] = struct{}{}

113 }

114 }

115

116 // Backup a single file by copying (never hardlinking, the file may change).

117 backupFile := func(path string) {

118 tmFile := time.Now()

119 srcpath := filepath.Join(srcDataDir, path)

120 dstpath := filepath.Join(dstDataDir, path)

121

122 sf, err := os.Open(srcpath)

123 if err != nil {

124 xerrx("open source file (not backed up)", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

125 return

126 }

127 defer sf.Close()

128

129 ensureDestDir(dstpath)

130 df, err := os.OpenFile(dstpath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0660)

131 if err != nil {

132 xerrx("creating destination file (not backed up)", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

133 return

134 }

135 defer func() {

136 if df != nil {

137 df.Close()

138 }

139 }()

140 if _, err := io.Copy(df, sf); err != nil {

141 xerrx("copying file (not backed up properly)", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

142 return

143 }

144 err = df.Close()

145 df = nil

146 if err != nil {

147 xerrx("closing destination file (not backed up properly)", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

148 return

149 }

150 xvlog("backed up file", slog.String("path", path), slog.Duration("duration", time.Since(tmFile)))

151 }

152

153 // Back up the files in a directory (by copying).

154 backupDir := func(dir string) {

155 tmDir := time.Now()

156 srcdir := filepath.Join(srcDataDir, dir)

157 dstdir := filepath.Join(dstDataDir, dir)

158 err := filepath.WalkDir(srcdir, func(srcpath string, d fs.DirEntry, err error) error {

159 if err != nil {

160 xerrx("walking file (not backed up)", err, slog.String("srcpath", srcpath))

161 return nil

162 }

163 if d.IsDir() {

164 return nil

165 }

166 backupFile(srcpath[len(srcDataDir)+1:])

167 return nil

168 })

169 if err != nil {

170 xerrx("copying directory (not backed up properly)", err,

171 slog.String("srcdir", srcdir),

172 slog.String("dstdir", dstdir),

173 slog.Duration("duration", time.Since(tmDir)))

174 return

175 }

176 xvlog("backed up directory", slog.String("dir", dir), slog.Duration("duration", time.Since(tmDir)))

177 }

178

179 // Backup a database by copying it in a readonly transaction.

180 // Always logs on error, so caller doesn't have to, but also returns the error so

181 // callers can see result.

182 backupDB := func(db *bstore.DB, path string) (rerr error) {

183 defer func() {

184 if rerr != nil {

185 xerrx("backing up database", rerr, slog.String("path", path))

186 }

187 }()

188

189 tmDB := time.Now()

190

191 dstpath := filepath.Join(dstDataDir, path)

192 ensureDestDir(dstpath)

193 df, err := os.OpenFile(dstpath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0660)

194 if err != nil {

195 return fmt.Errorf("creating destination file: %v", err)

196 }

197 defer func() {

198 if df != nil {

199 df.Close()

200 }

201 }()

202 err = db.Read(ctx, func(tx *bstore.Tx) error {

203 // Using regular WriteTo seems fine, and fast. It just copies pages.

204 //

205 // bolt.Compact is slower, it writes all key/value pairs, building up new data

206 // structures. My compacted test database was ~60% of original size. Lz4 on the

207 // uncompacted database got it to 14%. Lz4 on the compacted database got it to 13%.

208 // Backups are likely archived somewhere with compression, so we don't compact.

209 //

210 // Tests with WriteTo and os.O_DIRECT were slower than without O_DIRECT, but

211 // probably because everything fit in the page cache. It may be better to use

212 // O_DIRECT when copying many large or inactive databases.

213 _, err := tx.WriteTo(df)

214 return err

215 })

216 if err != nil {

217 return fmt.Errorf("copying database: %v", err)

218 }

219 err = df.Close()

220 df = nil

221 if err != nil {

222 return fmt.Errorf("closing destination database after copy: %v", err)

223 }

224 xvlog("backed up database file", slog.String("path", path), slog.Duration("duration", time.Since(tmDB)))

225 return nil

226 }

227

228 // Try to create a hardlink. Fall back to copying the file (e.g. when on different file system).

229 warnedHardlink := false // We warn once about failing to hardlink.

230 linkOrCopy := func(srcpath, dstpath string) (bool, error) {

231 ensureDestDir(dstpath)

232

233 if err := os.Link(srcpath, dstpath); err == nil {

234 return true, nil

235 } else if os.IsNotExist(err) {

236 // No point in trying with regular copy, we would warn twice.

237 return false, err

238 } else if !warnedHardlink {

239 var hardlinkHint string

240 if runtime.GOOS == "linux" && errors.Is(err, syscall.EXDEV) {

241 hardlinkHint = " (hint: if running under systemd, ReadWritePaths in mox.service may cause multiple mountpoints; consider merging paths into a single parent directory to prevent cross-device/mountpoint hardlinks)"

242 }

243 xwarnx("creating hardlink to message failed, will be doing regular file copies and not warn again"+hardlinkHint, err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

244 warnedHardlink = true

245 }

246

247 // Fall back to copying.

248 sf, err := os.Open(srcpath)

249 if err != nil {

250 return false, fmt.Errorf("open source path %s: %v", srcpath, err)

251 }

252 defer func() {

253 err := sf.Close()

254 ctl.log.Check(err, "closing copied source file")

255 }()

256

257 df, err := os.OpenFile(dstpath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0660)

258 if err != nil {

259 return false, fmt.Errorf("create destination path %s: %v", dstpath, err)

260 }

261 defer func() {

262 if df != nil {

263 err := df.Close()

264 ctl.log.Check(err, "closing partial destination file")

265 }

266 }()

267 if _, err := io.Copy(df, sf); err != nil {

268 return false, fmt.Errorf("coping: %v", err)

269 }

270 err = df.Close()

271 df = nil

272 if err != nil {

273 return false, fmt.Errorf("closing destination file: %v", err)

274 }

275 return false, nil

276 }

277

278 // Start making the backup.

279 tmStart := time.Now()

280

281 ctl.log.Print("making backup", slog.String("destdir", dstDataDir))

282

283 err := os.MkdirAll(dstDataDir, 0770)

284 if err != nil {

285 xerrx("creating destination data directory", err)

286 }

287

288 if err := os.WriteFile(filepath.Join(dstDataDir, "moxversion"), []byte(moxvar.Version), 0660); err != nil {

289 xerrx("writing moxversion", err)

290 }

291 backupDB(store.AuthDB, "auth.db")

292 backupDB(dmarcdb.ReportsDB, "dmarcrpt.db")

293 backupDB(dmarcdb.EvalDB, "dmarceval.db")

294 backupDB(mtastsdb.DB, "mtasts.db")

295 backupDB(tlsrptdb.ReportDB, "tlsrpt.db")

296 backupDB(tlsrptdb.ResultDB, "tlsrptresult.db")

297 backupFile("receivedid.key")

298

299 // Acme directory is optional.

300 srcAcmeDir := filepath.Join(srcDataDir, "acme")

301 if _, err := os.Stat(srcAcmeDir); err == nil {

302 backupDir("acme")

303 } else if err != nil && !os.IsNotExist(err) {

304 xerrx("copying acme/", err)

305 }

306

307 // Copy the queue database and all message files.

308 backupQueue := func(path string) {

309 tmQueue := time.Now()

310

311 if err := backupDB(queue.DB, path); err != nil {

312 xerrx("queue not backed up", err, slog.String("path", path), slog.Duration("duration", time.Since(tmQueue)))

313 return

314 }

315

316 dstdbpath := filepath.Join(dstDataDir, path)

317 opts := bstore.Options{MustExist: true, RegisterLogger: ctl.log.Logger}

318 db, err := bstore.Open(ctx, dstdbpath, &opts, queue.DBTypes...)

319 if err != nil {

320 xerrx("open copied queue database", err, slog.String("dstpath", dstdbpath), slog.Duration("duration", time.Since(tmQueue)))

321 return

322 }

323

324 defer func() {

325 if db != nil {

326 err := db.Close()

327 ctl.log.Check(err, "closing new queue db")

328 }

329 }()

330

331 // Link/copy known message files. Warn if files are missing or unexpected

332 // (though a message file could have been removed just now due to delivery, or a

333 // new message may have been queued).

334 tmMsgs := time.Now()

335 seen := map[string]struct{}{}

336 var nlinked, ncopied int

337 err = bstore.QueryDB[queue.Msg](ctx, db).ForEach(func(m queue.Msg) error {

338 mp := store.MessagePath(m.ID)

339 seen[mp] = struct{}{}

340 srcpath := filepath.Join(srcDataDir, "queue", mp)

341 dstpath := filepath.Join(dstDataDir, "queue", mp)

342 if linked, err := linkOrCopy(srcpath, dstpath); err != nil {

343 xerrx("linking/copying queue message", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

344 } else if linked {

345 nlinked++

346 } else {

347 ncopied++

348 }

349 return nil

350 })

351 if err != nil {

352 xerrx("processing queue messages (not backed up properly)", err, slog.Duration("duration", time.Since(tmMsgs)))

353 } else {

354 xvlog("queue message files linked/copied",

355 slog.Int("linked", nlinked),

356 slog.Int("copied", ncopied),

357 slog.Duration("duration", time.Since(tmMsgs)))

358 }

359

360 // Read through all files in queue directory and warn about anything we haven't handled yet.

361 tmWalk := time.Now()

362 srcqdir := filepath.Join(srcDataDir, "queue")

363 err = filepath.WalkDir(srcqdir, func(srcqpath string, d fs.DirEntry, err error) error {

364 if err != nil {

365 xerrx("walking files in queue", err, slog.String("srcpath", srcqpath))

366 return nil

367 }

368 if d.IsDir() {

369 return nil

370 }

371 p := srcqpath[len(srcqdir)+1:]

372 if _, ok := seen[p]; ok {

373 return nil

374 }

375 if p == "index.db" {

376 return nil

377 }

378 qp := filepath.Join("queue", p)

379 xwarnx("backing up unrecognized file in queue directory", nil, slog.String("path", qp))

380 backupFile(qp)

381 return nil

382 })

383 if err != nil {

384 xerrx("walking queue directory (not backed up properly)", err, slog.String("dir", "queue"), slog.Duration("duration", time.Since(tmWalk)))

385 } else {

386 xvlog("walked queue directory", slog.Duration("duration", time.Since(tmWalk)))

387 }

388

389 xvlog("queue backed finished", slog.Duration("duration", time.Since(tmQueue)))

390 }

391 backupQueue(filepath.FromSlash("queue/index.db"))

392

393 backupAccount := func(acc *store.Account) {

394 defer acc.Close()

395

396 tmAccount := time.Now()

397

398 // Copy database file.

399 dbpath := filepath.Join("accounts", acc.Name, "index.db")

400 err := backupDB(acc.DB, dbpath)

401 if err != nil {

402 xerrx("copying account database", err, slog.String("path", dbpath), slog.Duration("duration", time.Since(tmAccount)))

403 }

404

405 // todo: should document/check not taking a rlock on account.

406

407 // Copy junkfilter files, if configured.

408 if jf, _, err := acc.OpenJunkFilter(ctx, ctl.log); err != nil {

409 if !errors.Is(err, store.ErrNoJunkFilter) {

410 xerrx("opening junk filter for account (not backed up)", err)

411 }

412 } else {

413 db := jf.DB()

414 jfpath := filepath.Join("accounts", acc.Name, "junkfilter.db")

415 backupDB(db, jfpath)

416 bloompath := filepath.Join("accounts", acc.Name, "junkfilter.bloom")

417 backupFile(bloompath)

418 db = nil

419 err := jf.Close()

420 ctl.log.Check(err, "closing junkfilter")

421 }

422

423 dstdbpath := filepath.Join(dstDataDir, dbpath)

424 opts := bstore.Options{MustExist: true, RegisterLogger: ctl.log.Logger}

425 db, err := bstore.Open(ctx, dstdbpath, &opts, store.DBTypes...)

426 if err != nil {

427 xerrx("open copied account database", err, slog.String("dstpath", dstdbpath), slog.Duration("duration", time.Since(tmAccount)))

428 return

429 }

430

431 defer func() {

432 if db != nil {

433 err := db.Close()

434 ctl.log.Check(err, "close account database")

435 }

436 }()

437

438 // Link/copy known message files. Warn if files are missing or unexpected (though a

439 // message file could have been added just now due to delivery, or a message have

440 // been removed).

441 tmMsgs := time.Now()

442 seen := map[string]struct{}{}

443 var nlinked, ncopied int

444 err = bstore.QueryDB[store.Message](ctx, db).FilterEqual("Expunged", false).ForEach(func(m store.Message) error {

445 mp := store.MessagePath(m.ID)

446 seen[mp] = struct{}{}

447 amp := filepath.Join("accounts", acc.Name, "msg", mp)

448 srcpath := filepath.Join(srcDataDir, amp)

449 dstpath := filepath.Join(dstDataDir, amp)

450 if linked, err := linkOrCopy(srcpath, dstpath); err != nil {

451 xerrx("linking/copying account message", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

452 } else if linked {

453 nlinked++

454 } else {

455 ncopied++

456 }

457 return nil

458 })

459 if err != nil {

460 xerrx("processing account messages (not backed up properly)", err, slog.Duration("duration", time.Since(tmMsgs)))

461 } else {

462 xvlog("account message files linked/copied",

463 slog.Int("linked", nlinked),

464 slog.Int("copied", ncopied),

465 slog.Duration("duration", time.Since(tmMsgs)))

466 }

467

468 // Read through all files in account directory and warn about anything we haven't handled yet.

469 tmWalk := time.Now()

470 srcadir := filepath.Join(srcDataDir, "accounts", acc.Name)

471 err = filepath.WalkDir(srcadir, func(srcapath string, d fs.DirEntry, err error) error {

472 if err != nil {

473 xerrx("walking files in account", err, slog.String("srcpath", srcapath))

474 return nil

475 }

476 if d.IsDir() {

477 return nil

478 }

479 p := srcapath[len(srcadir)+1:]

480 l := strings.Split(p, string(filepath.Separator))

481 if l[0] == "msg" {

482 mp := filepath.Join(l[1:]...)

483 if _, ok := seen[mp]; ok {

484 return nil

485 }

486 }

487 switch p {

488 case "index.db", "junkfilter.db", "junkfilter.bloom":

489 return nil

490 }

491 ap := filepath.Join("accounts", acc.Name, p)

492 if strings.HasPrefix(p, "msg"+string(filepath.Separator)) {

493 xwarnx("backing up unrecognized file in account message directory (should be moved away)", nil, slog.String("path", ap))

494 } else {

495 xwarnx("backing up unrecognized file in account directory", nil, slog.String("path", ap))

496 }

497 backupFile(ap)

498 return nil

499 })

500 if err != nil {

501 xerrx("walking account directory (not backed up properly)", err, slog.String("srcdir", srcadir), slog.Duration("duration", time.Since(tmWalk)))

502 } else {

503 xvlog("walked account directory", slog.Duration("duration", time.Since(tmWalk)))

504 }

505

506 xvlog("account backup finished", slog.String("dir", filepath.Join("accounts", acc.Name)), slog.Duration("duration", time.Since(tmAccount)))

507 }

508

509 // For each configured account, open it, make a copy of the database and

510 // hardlink/copy the messages. We track the accounts we handled, and skip the

511 // account directories when handling "all other files" below.

512 accounts := map[string]struct{}{}

513 for _, accName := range mox.Conf.Accounts() {

514 acc, err := store.OpenAccount(ctl.log, accName)

515 if err != nil {

516 xerrx("opening account for copying (will try to copy as regular files later)", err, slog.String("account", accName))

517 continue

518 }

519 accounts[accName] = struct{}{}

520 backupAccount(acc)

521 }

522

523 // Copy all other files, that aren't part of the known files, databases, queue or accounts.

524 tmWalk := time.Now()

525 err = filepath.WalkDir(srcDataDir, func(srcpath string, d fs.DirEntry, err error) error {

526 if err != nil {

527 xerrx("walking path", err, slog.String("path", srcpath))

528 return nil

529 }

530

531 if srcpath == srcDataDir {

532 return nil

533 }

534 p := srcpath[len(srcDataDir)+1:]

535 if p == "queue" || p == "acme" || p == "tmp" {

536 return fs.SkipDir

537 }

538 l := strings.Split(p, string(filepath.Separator))

539 if len(l) >= 2 && l[0] == "accounts" {

540 name := l[1]

541 if _, ok := accounts[name]; ok {

542 return fs.SkipDir

543 }

544 }

545

546 // Only files are explicitly backed up.

547 if d.IsDir() {

548 return nil

549 }

550

551 switch p {

552 case "auth.db", "dmarcrpt.db", "dmarceval.db", "mtasts.db", "tlsrpt.db", "tlsrptresult.db", "receivedid.key", "ctl":

553 // Already handled.

554 return nil

555 case "lastknownversion": // Optional file, not yet handled.

556 default:

557 xwarnx("backing up unrecognized file", nil, slog.String("path", p))

558 }

559 backupFile(p)

560 return nil

561 })

562 if err != nil {

563 xerrx("walking other files (not backed up properly)", err, slog.Duration("duration", time.Since(tmWalk)))

564 } else {

565 xvlog("walking other files finished", slog.Duration("duration", time.Since(tmWalk)))

566 }

567

568 xvlog("backup finished", slog.Duration("duration", time.Since(tmStart)))

569

570 writer.xclose()

571

572 if incomplete {

573 ctl.xwrite("errors were encountered during backup")

574 } else {

575 ctl.xwriteok()

576 }

577}

578