leveldb.go 24 KB

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