leveldb.go 25 KB

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