leveldb.go 24 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051
  1. // Copyright (C) 2014 The Syncthing Authors.
  2. //
  3. // This Source Code Form is subject to the terms of the Mozilla Public
  4. // License, v. 2.0. If a copy of the MPL was not distributed with this file,
  5. // You can obtain one at http://mozilla.org/MPL/2.0/.
  6. //go:generate -command genxdr go run ../../Godeps/_workspace/src/github.com/calmh/xdr/cmd/genxdr/main.go
  7. //go:generate genxdr -o leveldb_xdr.go leveldb.go
  8. package db
  9. import (
  10. "bytes"
  11. "fmt"
  12. "runtime"
  13. "sort"
  14. "github.com/syncthing/protocol"
  15. "github.com/syncthing/syncthing/internal/sync"
  16. "github.com/syndtr/goleveldb/leveldb"
  17. "github.com/syndtr/goleveldb/leveldb/iterator"
  18. "github.com/syndtr/goleveldb/leveldb/opt"
  19. "github.com/syndtr/goleveldb/leveldb/util"
  20. )
  21. var (
  22. clockTick int64
  23. clockMut sync.Mutex = sync.NewMutex()
  24. )
  25. func clock(v int64) int64 {
  26. clockMut.Lock()
  27. defer clockMut.Unlock()
  28. if v > clockTick {
  29. clockTick = v + 1
  30. } else {
  31. clockTick++
  32. }
  33. return clockTick
  34. }
  35. const (
  36. KeyTypeDevice = iota
  37. KeyTypeGlobal
  38. KeyTypeBlock
  39. KeyTypeDeviceStatistic
  40. KeyTypeFolderStatistic
  41. )
  42. type fileVersion struct {
  43. version protocol.Vector
  44. device []byte
  45. }
  46. type versionList struct {
  47. versions []fileVersion
  48. }
  49. func (l versionList) String() string {
  50. var b bytes.Buffer
  51. var id protocol.DeviceID
  52. b.WriteString("{")
  53. for i, v := range l.versions {
  54. if i > 0 {
  55. b.WriteString(", ")
  56. }
  57. copy(id[:], v.device)
  58. fmt.Fprintf(&b, "{%d, %v}", v.version, id)
  59. }
  60. b.WriteString("}")
  61. return b.String()
  62. }
  63. type fileList []protocol.FileInfo
  64. func (l fileList) Len() int {
  65. return len(l)
  66. }
  67. func (l fileList) Swap(a, b int) {
  68. l[a], l[b] = l[b], l[a]
  69. }
  70. func (l fileList) Less(a, b int) bool {
  71. return l[a].Name < l[b].Name
  72. }
  73. type dbReader interface {
  74. Get([]byte, *opt.ReadOptions) ([]byte, error)
  75. }
  76. type dbWriter interface {
  77. Put([]byte, []byte)
  78. Delete([]byte)
  79. }
  80. // Flush batches to disk when they contain this many records.
  81. const batchFlushSize = 64
  82. // deviceKey returns a byte slice encoding the following information:
  83. // keyTypeDevice (1 byte)
  84. // folder (64 bytes)
  85. // device (32 bytes)
  86. // name (variable size)
  87. func deviceKey(folder, device, file []byte) []byte {
  88. k := make([]byte, 1+64+32+len(file))
  89. k[0] = KeyTypeDevice
  90. if len(folder) > 64 {
  91. panic("folder name too long")
  92. }
  93. copy(k[1:], []byte(folder))
  94. copy(k[1+64:], device[:])
  95. copy(k[1+64+32:], []byte(file))
  96. return k
  97. }
  98. func deviceKeyName(key []byte) []byte {
  99. return key[1+64+32:]
  100. }
  101. func deviceKeyFolder(key []byte) []byte {
  102. folder := key[1 : 1+64]
  103. izero := bytes.IndexByte(folder, 0)
  104. if izero < 0 {
  105. return folder
  106. }
  107. return folder[:izero]
  108. }
  109. func deviceKeyDevice(key []byte) []byte {
  110. return key[1+64 : 1+64+32]
  111. }
  112. // globalKey returns a byte slice encoding the following information:
  113. // keyTypeGlobal (1 byte)
  114. // folder (64 bytes)
  115. // name (variable size)
  116. func globalKey(folder, file []byte) []byte {
  117. k := make([]byte, 1+64+len(file))
  118. k[0] = KeyTypeGlobal
  119. if len(folder) > 64 {
  120. panic("folder name too long")
  121. }
  122. copy(k[1:], []byte(folder))
  123. copy(k[1+64:], []byte(file))
  124. return k
  125. }
  126. func globalKeyName(key []byte) []byte {
  127. return key[1+64:]
  128. }
  129. func globalKeyFolder(key []byte) []byte {
  130. folder := key[1 : 1+64]
  131. izero := bytes.IndexByte(folder, 0)
  132. if izero < 0 {
  133. return folder
  134. }
  135. return folder[:izero]
  136. }
  137. type deletionHandler func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64
  138. func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, deleteFn deletionHandler) int64 {
  139. runtime.GC()
  140. sort.Sort(fileList(fs)) // sort list on name, same as in the database
  141. start := deviceKey(folder, device, nil) // before all folder/device files
  142. limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
  143. batch := new(leveldb.Batch)
  144. if debugDB {
  145. l.Debugf("new batch %p", batch)
  146. }
  147. snap, err := db.GetSnapshot()
  148. if err != nil {
  149. panic(err)
  150. }
  151. if debugDB {
  152. l.Debugf("created snapshot %p", snap)
  153. }
  154. defer func() {
  155. if debugDB {
  156. l.Debugf("close snapshot %p", snap)
  157. }
  158. snap.Release()
  159. }()
  160. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  161. defer dbi.Release()
  162. moreDb := dbi.Next()
  163. fsi := 0
  164. var maxLocalVer int64
  165. for {
  166. var newName, oldName []byte
  167. moreFs := fsi < len(fs)
  168. if !moreDb && !moreFs {
  169. break
  170. }
  171. if moreFs {
  172. newName = []byte(fs[fsi].Name)
  173. }
  174. if moreDb {
  175. oldName = deviceKeyName(dbi.Key())
  176. }
  177. cmp := bytes.Compare(newName, oldName)
  178. if debugDB {
  179. l.Debugf("generic replace; folder=%q device=%v moreFs=%v moreDb=%v cmp=%d newName=%q oldName=%q", folder, protocol.DeviceIDFromBytes(device), moreFs, moreDb, cmp, newName, oldName)
  180. }
  181. switch {
  182. case moreFs && (!moreDb || cmp == -1):
  183. if debugDB {
  184. l.Debugln("generic replace; missing - insert")
  185. }
  186. // Database is missing this file. Insert it.
  187. if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer {
  188. maxLocalVer = lv
  189. }
  190. if fs[fsi].IsInvalid() {
  191. ldbRemoveFromGlobal(snap, batch, folder, device, newName)
  192. } else {
  193. ldbUpdateGlobal(snap, batch, folder, device, newName, fs[fsi].Version)
  194. }
  195. fsi++
  196. case moreFs && moreDb && cmp == 0:
  197. // File exists on both sides - compare versions. We might get an
  198. // update with the same version and different flags if a device has
  199. // marked a file as invalid, so handle that too.
  200. if debugDB {
  201. l.Debugln("generic replace; exists - compare")
  202. }
  203. var ef FileInfoTruncated
  204. ef.UnmarshalXDR(dbi.Value())
  205. if !fs[fsi].Version.Equal(ef.Version) || fs[fsi].Flags != ef.Flags {
  206. if debugDB {
  207. l.Debugln("generic replace; differs - insert")
  208. }
  209. if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer {
  210. maxLocalVer = lv
  211. }
  212. if fs[fsi].IsInvalid() {
  213. ldbRemoveFromGlobal(snap, batch, folder, device, newName)
  214. } else {
  215. ldbUpdateGlobal(snap, batch, folder, device, newName, fs[fsi].Version)
  216. }
  217. } else if debugDB {
  218. l.Debugln("generic replace; equal - ignore")
  219. }
  220. fsi++
  221. moreDb = dbi.Next()
  222. case moreDb && (!moreFs || cmp == 1):
  223. if debugDB {
  224. l.Debugln("generic replace; exists - remove")
  225. }
  226. if lv := deleteFn(snap, batch, folder, device, oldName, dbi); lv > maxLocalVer {
  227. maxLocalVer = lv
  228. }
  229. moreDb = dbi.Next()
  230. }
  231. // Write out and reuse the batch every few records, to avoid the batch
  232. // growing too large and thus allocating unnecessarily much memory.
  233. if batch.Len() > batchFlushSize {
  234. if debugDB {
  235. l.Debugf("db.Write %p", batch)
  236. }
  237. err = db.Write(batch, nil)
  238. if err != nil {
  239. panic(err)
  240. }
  241. batch.Reset()
  242. }
  243. }
  244. if debugDB {
  245. l.Debugf("db.Write %p", batch)
  246. }
  247. err = db.Write(batch, nil)
  248. if err != nil {
  249. panic(err)
  250. }
  251. return maxLocalVer
  252. }
  253. func ldbReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) int64 {
  254. // TODO: Return the remaining maxLocalVer?
  255. return ldbGenericReplace(db, folder, device, fs, func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64 {
  256. // Database has a file that we are missing. Remove it.
  257. if debugDB {
  258. l.Debugf("delete; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name)
  259. }
  260. ldbRemoveFromGlobal(db, batch, folder, device, name)
  261. if debugDB {
  262. l.Debugf("batch.Delete %p %x", batch, dbi.Key())
  263. }
  264. batch.Delete(dbi.Key())
  265. return 0
  266. })
  267. }
  268. func ldbReplaceWithDelete(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, myID uint64) int64 {
  269. return ldbGenericReplace(db, folder, device, fs, func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64 {
  270. var tf FileInfoTruncated
  271. err := tf.UnmarshalXDR(dbi.Value())
  272. if err != nil {
  273. panic(err)
  274. }
  275. if !tf.IsDeleted() {
  276. if debugDB {
  277. l.Debugf("mark deleted; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name)
  278. }
  279. ts := clock(tf.LocalVersion)
  280. f := protocol.FileInfo{
  281. Name: tf.Name,
  282. Version: tf.Version.Update(myID),
  283. LocalVersion: ts,
  284. Flags: tf.Flags | protocol.FlagDeleted,
  285. Modified: tf.Modified,
  286. }
  287. bs, _ := f.MarshalXDR()
  288. if debugDB {
  289. l.Debugf("batch.Put %p %x", batch, dbi.Key())
  290. }
  291. batch.Put(dbi.Key(), bs)
  292. ldbUpdateGlobal(db, batch, folder, device, deviceKeyName(dbi.Key()), f.Version)
  293. return ts
  294. }
  295. return 0
  296. })
  297. }
  298. func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo) int64 {
  299. runtime.GC()
  300. batch := new(leveldb.Batch)
  301. if debugDB {
  302. l.Debugf("new batch %p", batch)
  303. }
  304. snap, err := db.GetSnapshot()
  305. if err != nil {
  306. panic(err)
  307. }
  308. if debugDB {
  309. l.Debugf("created snapshot %p", snap)
  310. }
  311. defer func() {
  312. if debugDB {
  313. l.Debugf("close snapshot %p", snap)
  314. }
  315. snap.Release()
  316. }()
  317. var maxLocalVer int64
  318. for _, f := range fs {
  319. name := []byte(f.Name)
  320. fk := deviceKey(folder, device, name)
  321. if debugDB {
  322. l.Debugf("snap.Get %p %x", snap, fk)
  323. }
  324. bs, err := snap.Get(fk, nil)
  325. if err == leveldb.ErrNotFound {
  326. if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer {
  327. maxLocalVer = lv
  328. }
  329. if f.IsInvalid() {
  330. ldbRemoveFromGlobal(snap, batch, folder, device, name)
  331. } else {
  332. ldbUpdateGlobal(snap, batch, folder, device, name, f.Version)
  333. }
  334. continue
  335. }
  336. var ef FileInfoTruncated
  337. err = ef.UnmarshalXDR(bs)
  338. if err != nil {
  339. panic(err)
  340. }
  341. // Flags might change without the version being bumped when we set the
  342. // invalid flag on an existing file.
  343. if !ef.Version.Equal(f.Version) || ef.Flags != f.Flags {
  344. if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer {
  345. maxLocalVer = lv
  346. }
  347. if f.IsInvalid() {
  348. ldbRemoveFromGlobal(snap, batch, folder, device, name)
  349. } else {
  350. ldbUpdateGlobal(snap, batch, folder, device, name, f.Version)
  351. }
  352. }
  353. // Write out and reuse the batch every few records, to avoid the batch
  354. // growing too large and thus allocating unnecessarily much memory.
  355. if batch.Len() > batchFlushSize {
  356. if debugDB {
  357. l.Debugf("db.Write %p", batch)
  358. }
  359. err = db.Write(batch, nil)
  360. if err != nil {
  361. panic(err)
  362. }
  363. batch.Reset()
  364. }
  365. }
  366. if debugDB {
  367. l.Debugf("db.Write %p", batch)
  368. }
  369. err = db.Write(batch, nil)
  370. if err != nil {
  371. panic(err)
  372. }
  373. return maxLocalVer
  374. }
  375. func ldbInsert(batch dbWriter, folder, device []byte, file protocol.FileInfo) int64 {
  376. if debugDB {
  377. l.Debugf("insert; folder=%q device=%v %v", folder, protocol.DeviceIDFromBytes(device), file)
  378. }
  379. if file.LocalVersion == 0 {
  380. file.LocalVersion = clock(0)
  381. }
  382. name := []byte(file.Name)
  383. nk := deviceKey(folder, device, name)
  384. if debugDB {
  385. l.Debugf("batch.Put %p %x", batch, nk)
  386. }
  387. batch.Put(nk, file.MustMarshalXDR())
  388. return file.LocalVersion
  389. }
  390. // ldbUpdateGlobal adds this device+version to the version list for the given
  391. // file. If the device is already present in the list, the version is updated.
  392. // If the file does not have an entry in the global list, it is created.
  393. func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device, file []byte, version protocol.Vector) bool {
  394. if debugDB {
  395. l.Debugf("update global; folder=%q device=%v file=%q version=%d", folder, protocol.DeviceIDFromBytes(device), file, version)
  396. }
  397. gk := globalKey(folder, file)
  398. svl, err := db.Get(gk, nil)
  399. if err != nil && err != leveldb.ErrNotFound {
  400. panic(err)
  401. }
  402. var fl versionList
  403. // Remove the device from the current version list
  404. if svl != nil {
  405. err = fl.UnmarshalXDR(svl)
  406. if err != nil {
  407. panic(err)
  408. }
  409. for i := range fl.versions {
  410. if bytes.Compare(fl.versions[i].device, device) == 0 {
  411. if fl.versions[i].version.Equal(version) {
  412. // No need to do anything
  413. return false
  414. }
  415. fl.versions = append(fl.versions[:i], fl.versions[i+1:]...)
  416. break
  417. }
  418. }
  419. }
  420. nv := fileVersion{
  421. device: device,
  422. version: version,
  423. }
  424. for i := range fl.versions {
  425. // We compare against ConcurrentLesser as well here because we need
  426. // to enforce a consistent ordering of versions even in the case of
  427. // conflicts.
  428. if comp := fl.versions[i].version.Compare(version); comp == protocol.Equal || comp == protocol.Lesser || comp == protocol.ConcurrentLesser {
  429. t := append(fl.versions, fileVersion{})
  430. copy(t[i+1:], t[i:])
  431. t[i] = nv
  432. fl.versions = t
  433. goto done
  434. }
  435. }
  436. fl.versions = append(fl.versions, nv)
  437. done:
  438. if debugDB {
  439. l.Debugf("batch.Put %p %x", batch, gk)
  440. l.Debugf("new global after update: %v", fl)
  441. }
  442. batch.Put(gk, fl.MustMarshalXDR())
  443. return true
  444. }
  445. // ldbRemoveFromGlobal removes the device from the global version list for the
  446. // given file. If the version list is empty after this, the file entry is
  447. // removed entirely.
  448. func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byte) {
  449. if debugDB {
  450. l.Debugf("remove from global; folder=%q device=%v file=%q", folder, protocol.DeviceIDFromBytes(device), file)
  451. }
  452. gk := globalKey(folder, file)
  453. svl, err := db.Get(gk, nil)
  454. if err != nil {
  455. // We might be called to "remove" a global version that doesn't exist
  456. // if the first update for the file is already marked invalid.
  457. return
  458. }
  459. var fl versionList
  460. err = fl.UnmarshalXDR(svl)
  461. if err != nil {
  462. panic(err)
  463. }
  464. for i := range fl.versions {
  465. if bytes.Compare(fl.versions[i].device, device) == 0 {
  466. fl.versions = append(fl.versions[:i], fl.versions[i+1:]...)
  467. break
  468. }
  469. }
  470. if len(fl.versions) == 0 {
  471. if debugDB {
  472. l.Debugf("batch.Delete %p %x", batch, gk)
  473. }
  474. batch.Delete(gk)
  475. } else {
  476. if debugDB {
  477. l.Debugf("batch.Put %p %x", batch, gk)
  478. l.Debugf("new global after remove: %v", fl)
  479. }
  480. batch.Put(gk, fl.MustMarshalXDR())
  481. }
  482. }
  483. func ldbWithHave(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterator) {
  484. start := deviceKey(folder, device, nil) // before all folder/device files
  485. limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
  486. snap, err := db.GetSnapshot()
  487. if err != nil {
  488. panic(err)
  489. }
  490. if debugDB {
  491. l.Debugf("created snapshot %p", snap)
  492. }
  493. defer func() {
  494. if debugDB {
  495. l.Debugf("close snapshot %p", snap)
  496. }
  497. snap.Release()
  498. }()
  499. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  500. defer dbi.Release()
  501. for dbi.Next() {
  502. f, err := unmarshalTrunc(dbi.Value(), truncate)
  503. if err != nil {
  504. panic(err)
  505. }
  506. if cont := fn(f); !cont {
  507. return
  508. }
  509. }
  510. }
  511. func ldbWithAllFolderTruncated(db *leveldb.DB, folder []byte, fn func(device []byte, f FileInfoTruncated) bool) {
  512. runtime.GC()
  513. start := deviceKey(folder, nil, nil) // before all folder/device files
  514. limit := deviceKey(folder, protocol.LocalDeviceID[:], []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
  515. snap, err := db.GetSnapshot()
  516. if err != nil {
  517. panic(err)
  518. }
  519. if debugDB {
  520. l.Debugf("created snapshot %p", snap)
  521. }
  522. defer func() {
  523. if debugDB {
  524. l.Debugf("close snapshot %p", snap)
  525. }
  526. snap.Release()
  527. }()
  528. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  529. defer dbi.Release()
  530. for dbi.Next() {
  531. device := deviceKeyDevice(dbi.Key())
  532. var f FileInfoTruncated
  533. err := f.UnmarshalXDR(dbi.Value())
  534. if err != nil {
  535. panic(err)
  536. }
  537. switch f.Name {
  538. case "", ".", "..", "/": // A few obviously invalid filenames
  539. l.Infof("Dropping invalid filename %q from database", f.Name)
  540. batch := new(leveldb.Batch)
  541. ldbRemoveFromGlobal(db, batch, folder, device, nil)
  542. batch.Delete(dbi.Key())
  543. db.Write(batch, nil)
  544. continue
  545. }
  546. if cont := fn(device, f); !cont {
  547. return
  548. }
  549. }
  550. }
  551. func ldbGet(db *leveldb.DB, folder, device, file []byte) (protocol.FileInfo, bool) {
  552. nk := deviceKey(folder, device, file)
  553. bs, err := db.Get(nk, nil)
  554. if err == leveldb.ErrNotFound {
  555. return protocol.FileInfo{}, false
  556. }
  557. if err != nil {
  558. panic(err)
  559. }
  560. var f protocol.FileInfo
  561. err = f.UnmarshalXDR(bs)
  562. if err != nil {
  563. panic(err)
  564. }
  565. return f, true
  566. }
  567. func ldbGetGlobal(db *leveldb.DB, folder, file []byte, truncate bool) (FileIntf, bool) {
  568. k := globalKey(folder, file)
  569. snap, err := db.GetSnapshot()
  570. if err != nil {
  571. panic(err)
  572. }
  573. if debugDB {
  574. l.Debugf("created snapshot %p", snap)
  575. }
  576. defer func() {
  577. if debugDB {
  578. l.Debugf("close snapshot %p", snap)
  579. }
  580. snap.Release()
  581. }()
  582. if debugDB {
  583. l.Debugf("snap.Get %p %x", snap, k)
  584. }
  585. bs, err := snap.Get(k, nil)
  586. if err == leveldb.ErrNotFound {
  587. return nil, false
  588. }
  589. if err != nil {
  590. panic(err)
  591. }
  592. var vl versionList
  593. err = vl.UnmarshalXDR(bs)
  594. if err != nil {
  595. panic(err)
  596. }
  597. if len(vl.versions) == 0 {
  598. l.Debugln(k)
  599. panic("no versions?")
  600. }
  601. k = deviceKey(folder, vl.versions[0].device, file)
  602. if debugDB {
  603. l.Debugf("snap.Get %p %x", snap, k)
  604. }
  605. bs, err = snap.Get(k, nil)
  606. if err != nil {
  607. panic(err)
  608. }
  609. fi, err := unmarshalTrunc(bs, truncate)
  610. if err != nil {
  611. panic(err)
  612. }
  613. return fi, true
  614. }
  615. func ldbWithGlobal(db *leveldb.DB, folder, prefix []byte, truncate bool, fn Iterator) {
  616. runtime.GC()
  617. snap, err := db.GetSnapshot()
  618. if err != nil {
  619. panic(err)
  620. }
  621. if debugDB {
  622. l.Debugf("created snapshot %p", snap)
  623. }
  624. defer func() {
  625. if debugDB {
  626. l.Debugf("close snapshot %p", snap)
  627. }
  628. snap.Release()
  629. }()
  630. dbi := snap.NewIterator(util.BytesPrefix(globalKey(folder, prefix)), nil)
  631. defer dbi.Release()
  632. for dbi.Next() {
  633. var vl versionList
  634. err := vl.UnmarshalXDR(dbi.Value())
  635. if err != nil {
  636. panic(err)
  637. }
  638. if len(vl.versions) == 0 {
  639. l.Debugln(dbi.Key())
  640. panic("no versions?")
  641. }
  642. name := globalKeyName(dbi.Key())
  643. fk := deviceKey(folder, vl.versions[0].device, name)
  644. if debugDB {
  645. l.Debugf("snap.Get %p %x", snap, fk)
  646. }
  647. bs, err := snap.Get(fk, nil)
  648. if err != nil {
  649. l.Debugf("folder: %q (%x)", folder, folder)
  650. l.Debugf("key: %q (%x)", dbi.Key(), dbi.Key())
  651. l.Debugf("vl: %v", vl)
  652. l.Debugf("vl.versions[0].device: %x", vl.versions[0].device)
  653. l.Debugf("name: %q (%x)", name, name)
  654. l.Debugf("fk: %q", fk)
  655. l.Debugf("fk: %x %x %x", fk[1:1+64], fk[1+64:1+64+32], fk[1+64+32:])
  656. panic(err)
  657. }
  658. f, err := unmarshalTrunc(bs, truncate)
  659. if err != nil {
  660. panic(err)
  661. }
  662. if cont := fn(f); !cont {
  663. return
  664. }
  665. }
  666. }
  667. func ldbAvailability(db *leveldb.DB, folder, file []byte) []protocol.DeviceID {
  668. k := globalKey(folder, file)
  669. bs, err := db.Get(k, nil)
  670. if err == leveldb.ErrNotFound {
  671. return nil
  672. }
  673. if err != nil {
  674. panic(err)
  675. }
  676. var vl versionList
  677. err = vl.UnmarshalXDR(bs)
  678. if err != nil {
  679. panic(err)
  680. }
  681. var devices []protocol.DeviceID
  682. for _, v := range vl.versions {
  683. if !v.version.Equal(vl.versions[0].version) {
  684. break
  685. }
  686. n := protocol.DeviceIDFromBytes(v.device)
  687. devices = append(devices, n)
  688. }
  689. return devices
  690. }
  691. func ldbWithNeed(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterator) {
  692. runtime.GC()
  693. start := globalKey(folder, nil)
  694. limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
  695. snap, err := db.GetSnapshot()
  696. if err != nil {
  697. panic(err)
  698. }
  699. if debugDB {
  700. l.Debugf("created snapshot %p", snap)
  701. }
  702. defer func() {
  703. if debugDB {
  704. l.Debugf("close snapshot %p", snap)
  705. }
  706. snap.Release()
  707. }()
  708. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  709. defer dbi.Release()
  710. nextFile:
  711. for dbi.Next() {
  712. var vl versionList
  713. err := vl.UnmarshalXDR(dbi.Value())
  714. if err != nil {
  715. panic(err)
  716. }
  717. if len(vl.versions) == 0 {
  718. l.Debugln(dbi.Key())
  719. panic("no versions?")
  720. }
  721. have := false // If we have the file, any version
  722. need := false // If we have a lower version of the file
  723. var haveVersion protocol.Vector
  724. for _, v := range vl.versions {
  725. if bytes.Compare(v.device, device) == 0 {
  726. have = true
  727. haveVersion = v.version
  728. // XXX: This marks Concurrent (i.e. conflicting) changes as
  729. // needs. Maybe we should do that, but it needs special
  730. // handling in the puller.
  731. need = !v.version.GreaterEqual(vl.versions[0].version)
  732. break
  733. }
  734. }
  735. if need || !have {
  736. name := globalKeyName(dbi.Key())
  737. needVersion := vl.versions[0].version
  738. nextVersion:
  739. for i := range vl.versions {
  740. if !vl.versions[i].version.Equal(needVersion) {
  741. // We haven't found a valid copy of the file with the needed version.
  742. continue nextFile
  743. }
  744. fk := deviceKey(folder, vl.versions[i].device, name)
  745. if debugDB {
  746. l.Debugf("snap.Get %p %x", snap, fk)
  747. }
  748. bs, err := snap.Get(fk, nil)
  749. if err != nil {
  750. var id protocol.DeviceID
  751. copy(id[:], device)
  752. l.Debugf("device: %v", id)
  753. l.Debugf("need: %v, have: %v", need, have)
  754. l.Debugf("key: %q (%x)", dbi.Key(), dbi.Key())
  755. l.Debugf("vl: %v", vl)
  756. l.Debugf("i: %v", i)
  757. l.Debugf("fk: %q (%x)", fk, fk)
  758. l.Debugf("name: %q (%x)", name, name)
  759. panic(err)
  760. }
  761. gf, err := unmarshalTrunc(bs, truncate)
  762. if err != nil {
  763. panic(err)
  764. }
  765. if gf.IsInvalid() {
  766. // The file is marked invalid for whatever reason, don't use it.
  767. continue nextVersion
  768. }
  769. if gf.IsDeleted() && !have {
  770. // We don't need deleted files that we don't have
  771. continue nextFile
  772. }
  773. if debugDB {
  774. l.Debugf("need folder=%q device=%v name=%q need=%v have=%v haveV=%d globalV=%d", folder, protocol.DeviceIDFromBytes(device), name, need, have, haveVersion, vl.versions[0].version)
  775. }
  776. if cont := fn(gf); !cont {
  777. return
  778. }
  779. // This file is handled, no need to look further in the version list
  780. continue nextFile
  781. }
  782. }
  783. }
  784. }
  785. func ldbListFolders(db *leveldb.DB) []string {
  786. runtime.GC()
  787. snap, err := db.GetSnapshot()
  788. if err != nil {
  789. panic(err)
  790. }
  791. if debugDB {
  792. l.Debugf("created snapshot %p", snap)
  793. }
  794. defer func() {
  795. if debugDB {
  796. l.Debugf("close snapshot %p", snap)
  797. }
  798. snap.Release()
  799. }()
  800. dbi := snap.NewIterator(util.BytesPrefix([]byte{KeyTypeGlobal}), nil)
  801. defer dbi.Release()
  802. folderExists := make(map[string]bool)
  803. for dbi.Next() {
  804. folder := string(globalKeyFolder(dbi.Key()))
  805. if !folderExists[folder] {
  806. folderExists[folder] = true
  807. }
  808. }
  809. folders := make([]string, 0, len(folderExists))
  810. for k := range folderExists {
  811. folders = append(folders, k)
  812. }
  813. sort.Strings(folders)
  814. return folders
  815. }
  816. func ldbDropFolder(db *leveldb.DB, folder []byte) {
  817. runtime.GC()
  818. snap, err := db.GetSnapshot()
  819. if err != nil {
  820. panic(err)
  821. }
  822. if debugDB {
  823. l.Debugf("created snapshot %p", snap)
  824. }
  825. defer func() {
  826. if debugDB {
  827. l.Debugf("close snapshot %p", snap)
  828. }
  829. snap.Release()
  830. }()
  831. // Remove all items related to the given folder from the device->file bucket
  832. dbi := snap.NewIterator(util.BytesPrefix([]byte{KeyTypeDevice}), nil)
  833. for dbi.Next() {
  834. itemFolder := deviceKeyFolder(dbi.Key())
  835. if bytes.Compare(folder, itemFolder) == 0 {
  836. db.Delete(dbi.Key(), nil)
  837. }
  838. }
  839. dbi.Release()
  840. // Remove all items related to the given folder from the global bucket
  841. dbi = snap.NewIterator(util.BytesPrefix([]byte{KeyTypeGlobal}), nil)
  842. for dbi.Next() {
  843. itemFolder := globalKeyFolder(dbi.Key())
  844. if bytes.Compare(folder, itemFolder) == 0 {
  845. db.Delete(dbi.Key(), nil)
  846. }
  847. }
  848. dbi.Release()
  849. }
  850. func unmarshalTrunc(bs []byte, truncate bool) (FileIntf, error) {
  851. if truncate {
  852. var tf FileInfoTruncated
  853. err := tf.UnmarshalXDR(bs)
  854. return tf, err
  855. } else {
  856. var tf protocol.FileInfo
  857. err := tf.UnmarshalXDR(bs)
  858. return tf, err
  859. }
  860. }
  861. func ldbCheckGlobals(db *leveldb.DB, folder []byte) {
  862. defer runtime.GC()
  863. snap, err := db.GetSnapshot()
  864. if err != nil {
  865. panic(err)
  866. }
  867. if debugDB {
  868. l.Debugf("created snapshot %p", snap)
  869. }
  870. defer func() {
  871. if debugDB {
  872. l.Debugf("close snapshot %p", snap)
  873. }
  874. snap.Release()
  875. }()
  876. start := globalKey(folder, nil)
  877. limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
  878. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  879. defer dbi.Release()
  880. batch := new(leveldb.Batch)
  881. if debugDB {
  882. l.Debugf("new batch %p", batch)
  883. }
  884. for dbi.Next() {
  885. gk := dbi.Key()
  886. var vl versionList
  887. err := vl.UnmarshalXDR(dbi.Value())
  888. if err != nil {
  889. panic(err)
  890. }
  891. // Check the global version list for consistency. An issue in previous
  892. // versions of goleveldb could result in reordered writes so that
  893. // there are global entries pointing to no longer existing files. Here
  894. // we find those and clear them out.
  895. name := globalKeyName(gk)
  896. var newVL versionList
  897. for _, version := range vl.versions {
  898. fk := deviceKey(folder, version.device, name)
  899. if debugDB {
  900. l.Debugf("snap.Get %p %x", snap, fk)
  901. }
  902. _, err := snap.Get(fk, nil)
  903. if err == leveldb.ErrNotFound {
  904. continue
  905. }
  906. if err != nil {
  907. panic(err)
  908. }
  909. newVL.versions = append(newVL.versions, version)
  910. }
  911. if len(newVL.versions) != len(vl.versions) {
  912. l.Infof("db repair: rewriting global version list for %x %x", gk[1:1+64], gk[1+64:])
  913. batch.Put(dbi.Key(), newVL.MustMarshalXDR())
  914. }
  915. }
  916. if debugDB {
  917. l.Infoln("db check completed for %q", folder)
  918. }
  919. db.Write(batch, nil)
  920. }