model.go 49 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812
  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. package model
  7. import (
  8. "bufio"
  9. "crypto/tls"
  10. "encoding/json"
  11. "errors"
  12. "fmt"
  13. "io"
  14. "io/ioutil"
  15. "net"
  16. "os"
  17. "path/filepath"
  18. "reflect"
  19. "runtime"
  20. "strings"
  21. stdsync "sync"
  22. "time"
  23. "github.com/syncthing/protocol"
  24. "github.com/syncthing/syncthing/internal/config"
  25. "github.com/syncthing/syncthing/internal/db"
  26. "github.com/syncthing/syncthing/internal/events"
  27. "github.com/syncthing/syncthing/internal/ignore"
  28. "github.com/syncthing/syncthing/internal/osutil"
  29. "github.com/syncthing/syncthing/internal/scanner"
  30. "github.com/syncthing/syncthing/internal/stats"
  31. "github.com/syncthing/syncthing/internal/symlinks"
  32. "github.com/syncthing/syncthing/internal/sync"
  33. "github.com/syncthing/syncthing/internal/versioner"
  34. "github.com/syndtr/goleveldb/leveldb"
  35. "github.com/thejerf/suture"
  36. )
  37. // How many files to send in each Index/IndexUpdate message.
  38. const (
  39. indexTargetSize = 250 * 1024 // Aim for making index messages no larger than 250 KiB (uncompressed)
  40. indexPerFileSize = 250 // Each FileInfo is approximately this big, in bytes, excluding BlockInfos
  41. indexPerBlockSize = 40 // Each BlockInfo is approximately this big
  42. indexBatchSize = 1000 // Either way, don't include more files than this
  43. reqValidationTime = time.Hour // How long to cache validation entries for Request messages
  44. reqValidationCacheSize = 1000 // How many entries to aim for in the validation cache size
  45. )
  46. type service interface {
  47. Serve()
  48. Stop()
  49. Jobs() ([]string, []string) // In progress, Queued
  50. BringToFront(string)
  51. DelayScan(d time.Duration)
  52. IndexUpdated() // Remote index was updated notification
  53. setState(state folderState)
  54. setError(err error)
  55. clearError()
  56. getState() (folderState, time.Time, error)
  57. }
  58. type Model struct {
  59. *suture.Supervisor
  60. cfg *config.Wrapper
  61. db *leveldb.DB
  62. finder *db.BlockFinder
  63. progressEmitter *ProgressEmitter
  64. id protocol.DeviceID
  65. shortID uint64
  66. deviceName string
  67. clientName string
  68. clientVersion string
  69. folderCfgs map[string]config.FolderConfiguration // folder -> cfg
  70. folderFiles map[string]*db.FileSet // folder -> files
  71. folderDevices map[string][]protocol.DeviceID // folder -> deviceIDs
  72. deviceFolders map[protocol.DeviceID][]string // deviceID -> folders
  73. deviceStatRefs map[protocol.DeviceID]*stats.DeviceStatisticsReference // deviceID -> statsRef
  74. folderIgnores map[string]*ignore.Matcher // folder -> matcher object
  75. folderRunners map[string]service // folder -> puller or scanner
  76. folderStatRefs map[string]*stats.FolderStatisticsReference // folder -> statsRef
  77. fmut sync.RWMutex // protects the above
  78. protoConn map[protocol.DeviceID]protocol.Connection
  79. rawConn map[protocol.DeviceID]io.Closer
  80. deviceVer map[protocol.DeviceID]string
  81. pmut sync.RWMutex // protects protoConn and rawConn
  82. addedFolder bool
  83. started bool
  84. reqValidationCache map[string]time.Time // folder / file name => time when confirmed to exist
  85. rvmut sync.RWMutex // protects reqValidationCache
  86. }
  87. var (
  88. symlinkWarning = stdsync.Once{}
  89. )
  90. // NewModel creates and starts a new model. The model starts in read-only mode,
  91. // where it sends index information to connected peers and responds to requests
  92. // for file data without altering the local folder in any way.
  93. func NewModel(cfg *config.Wrapper, id protocol.DeviceID, deviceName, clientName, clientVersion string, ldb *leveldb.DB) *Model {
  94. m := &Model{
  95. Supervisor: suture.New("model", suture.Spec{
  96. Log: func(line string) {
  97. if debug {
  98. l.Debugln(line)
  99. }
  100. },
  101. }),
  102. cfg: cfg,
  103. db: ldb,
  104. finder: db.NewBlockFinder(ldb, cfg),
  105. progressEmitter: NewProgressEmitter(cfg),
  106. id: id,
  107. shortID: id.Short(),
  108. deviceName: deviceName,
  109. clientName: clientName,
  110. clientVersion: clientVersion,
  111. folderCfgs: make(map[string]config.FolderConfiguration),
  112. folderFiles: make(map[string]*db.FileSet),
  113. folderDevices: make(map[string][]protocol.DeviceID),
  114. deviceFolders: make(map[protocol.DeviceID][]string),
  115. deviceStatRefs: make(map[protocol.DeviceID]*stats.DeviceStatisticsReference),
  116. folderIgnores: make(map[string]*ignore.Matcher),
  117. folderRunners: make(map[string]service),
  118. folderStatRefs: make(map[string]*stats.FolderStatisticsReference),
  119. protoConn: make(map[protocol.DeviceID]protocol.Connection),
  120. rawConn: make(map[protocol.DeviceID]io.Closer),
  121. deviceVer: make(map[protocol.DeviceID]string),
  122. reqValidationCache: make(map[string]time.Time),
  123. fmut: sync.NewRWMutex(),
  124. pmut: sync.NewRWMutex(),
  125. rvmut: sync.NewRWMutex(),
  126. }
  127. if cfg.Options().ProgressUpdateIntervalS > -1 {
  128. go m.progressEmitter.Serve()
  129. }
  130. return m
  131. }
  132. // StartDeadlockDetector starts a deadlock detector on the models locks which
  133. // causes panics in case the locks cannot be acquired in the given timeout
  134. // period.
  135. func (m *Model) StartDeadlockDetector(timeout time.Duration) {
  136. l.Infof("Starting deadlock detector with %v timeout", timeout)
  137. deadlockDetect(m.fmut, timeout)
  138. deadlockDetect(m.pmut, timeout)
  139. }
  140. // StartFolderRW starts read/write processing on the current model. When in
  141. // read/write mode the model will attempt to keep in sync with the cluster by
  142. // pulling needed files from peer devices.
  143. func (m *Model) StartFolderRW(folder string) {
  144. m.fmut.Lock()
  145. cfg, ok := m.folderCfgs[folder]
  146. if !ok {
  147. panic("cannot start nonexistent folder " + folder)
  148. }
  149. _, ok = m.folderRunners[folder]
  150. if ok {
  151. panic("cannot start already running folder " + folder)
  152. }
  153. p := newRWFolder(m, m.shortID, cfg)
  154. m.folderRunners[folder] = p
  155. m.fmut.Unlock()
  156. if len(cfg.Versioning.Type) > 0 {
  157. factory, ok := versioner.Factories[cfg.Versioning.Type]
  158. if !ok {
  159. l.Fatalf("Requested versioning type %q that does not exist", cfg.Versioning.Type)
  160. }
  161. versioner := factory(folder, cfg.Path(), cfg.Versioning.Params)
  162. if service, ok := versioner.(suture.Service); ok {
  163. // The versioner implements the suture.Service interface, so
  164. // expects to be run in the background in addition to being called
  165. // when files are going to be archived.
  166. m.Add(service)
  167. }
  168. p.versioner = versioner
  169. }
  170. go p.Serve()
  171. }
  172. // StartFolderRO starts read only processing on the current model. When in
  173. // read only mode the model will announce files to the cluster but not pull in
  174. // any external changes.
  175. func (m *Model) StartFolderRO(folder string) {
  176. m.fmut.Lock()
  177. cfg, ok := m.folderCfgs[folder]
  178. if !ok {
  179. panic("cannot start nonexistent folder " + folder)
  180. }
  181. _, ok = m.folderRunners[folder]
  182. if ok {
  183. panic("cannot start already running folder " + folder)
  184. }
  185. s := newROFolder(m, folder, time.Duration(cfg.RescanIntervalS)*time.Second)
  186. m.folderRunners[folder] = s
  187. m.fmut.Unlock()
  188. go s.Serve()
  189. }
  190. type ConnectionInfo struct {
  191. protocol.Statistics
  192. Address string
  193. ClientVersion string
  194. }
  195. func (info ConnectionInfo) MarshalJSON() ([]byte, error) {
  196. return json.Marshal(map[string]interface{}{
  197. "at": info.At,
  198. "inBytesTotal": info.InBytesTotal,
  199. "outBytesTotal": info.OutBytesTotal,
  200. "address": info.Address,
  201. "clientVersion": info.ClientVersion,
  202. })
  203. }
  204. // ConnectionStats returns a map with connection statistics for each connected device.
  205. func (m *Model) ConnectionStats() map[string]interface{} {
  206. type remoteAddrer interface {
  207. RemoteAddr() net.Addr
  208. }
  209. m.pmut.RLock()
  210. m.fmut.RLock()
  211. var res = make(map[string]interface{})
  212. conns := make(map[string]ConnectionInfo, len(m.protoConn))
  213. for device, conn := range m.protoConn {
  214. ci := ConnectionInfo{
  215. Statistics: conn.Statistics(),
  216. ClientVersion: m.deviceVer[device],
  217. }
  218. if nc, ok := m.rawConn[device].(remoteAddrer); ok {
  219. ci.Address = nc.RemoteAddr().String()
  220. }
  221. conns[device.String()] = ci
  222. }
  223. res["connections"] = conns
  224. m.fmut.RUnlock()
  225. m.pmut.RUnlock()
  226. in, out := protocol.TotalInOut()
  227. res["total"] = ConnectionInfo{
  228. Statistics: protocol.Statistics{
  229. At: time.Now(),
  230. InBytesTotal: in,
  231. OutBytesTotal: out,
  232. },
  233. }
  234. return res
  235. }
  236. // DeviceStatistics returns statistics about each device
  237. func (m *Model) DeviceStatistics() map[string]stats.DeviceStatistics {
  238. var res = make(map[string]stats.DeviceStatistics)
  239. for id := range m.cfg.Devices() {
  240. res[id.String()] = m.deviceStatRef(id).GetStatistics()
  241. }
  242. return res
  243. }
  244. // FolderStatistics returns statistics about each folder
  245. func (m *Model) FolderStatistics() map[string]stats.FolderStatistics {
  246. var res = make(map[string]stats.FolderStatistics)
  247. for id := range m.cfg.Folders() {
  248. res[id] = m.folderStatRef(id).GetStatistics()
  249. }
  250. return res
  251. }
  252. // Completion returns the completion status, in percent, for the given device
  253. // and folder.
  254. func (m *Model) Completion(device protocol.DeviceID, folder string) float64 {
  255. var tot int64
  256. m.fmut.RLock()
  257. rf, ok := m.folderFiles[folder]
  258. m.fmut.RUnlock()
  259. if !ok {
  260. return 0 // Folder doesn't exist, so we hardly have any of it
  261. }
  262. rf.WithGlobalTruncated(func(f db.FileIntf) bool {
  263. if !f.IsDeleted() {
  264. tot += f.Size()
  265. }
  266. return true
  267. })
  268. if tot == 0 {
  269. return 100 // Folder is empty, so we have all of it
  270. }
  271. var need int64
  272. rf.WithNeedTruncated(device, func(f db.FileIntf) bool {
  273. if !f.IsDeleted() {
  274. need += f.Size()
  275. }
  276. return true
  277. })
  278. res := 100 * (1 - float64(need)/float64(tot))
  279. if debug {
  280. l.Debugf("%v Completion(%s, %q): %f (%d / %d)", m, device, folder, res, need, tot)
  281. }
  282. return res
  283. }
  284. func sizeOf(fs []protocol.FileInfo) (files, deleted int, bytes int64) {
  285. for _, f := range fs {
  286. fs, de, by := sizeOfFile(f)
  287. files += fs
  288. deleted += de
  289. bytes += by
  290. }
  291. return
  292. }
  293. func sizeOfFile(f db.FileIntf) (files, deleted int, bytes int64) {
  294. if !f.IsDeleted() {
  295. files++
  296. } else {
  297. deleted++
  298. }
  299. bytes += f.Size()
  300. return
  301. }
  302. // GlobalSize returns the number of files, deleted files and total bytes for all
  303. // files in the global model.
  304. func (m *Model) GlobalSize(folder string) (nfiles, deleted int, bytes int64) {
  305. m.fmut.RLock()
  306. defer m.fmut.RUnlock()
  307. if rf, ok := m.folderFiles[folder]; ok {
  308. rf.WithGlobalTruncated(func(f db.FileIntf) bool {
  309. fs, de, by := sizeOfFile(f)
  310. nfiles += fs
  311. deleted += de
  312. bytes += by
  313. return true
  314. })
  315. }
  316. return
  317. }
  318. // LocalSize returns the number of files, deleted files and total bytes for all
  319. // files in the local folder.
  320. func (m *Model) LocalSize(folder string) (nfiles, deleted int, bytes int64) {
  321. m.fmut.RLock()
  322. defer m.fmut.RUnlock()
  323. if rf, ok := m.folderFiles[folder]; ok {
  324. rf.WithHaveTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  325. if f.IsInvalid() {
  326. return true
  327. }
  328. fs, de, by := sizeOfFile(f)
  329. nfiles += fs
  330. deleted += de
  331. bytes += by
  332. return true
  333. })
  334. }
  335. return
  336. }
  337. // NeedSize returns the number and total size of currently needed files.
  338. func (m *Model) NeedSize(folder string) (nfiles int, bytes int64) {
  339. m.fmut.RLock()
  340. defer m.fmut.RUnlock()
  341. if rf, ok := m.folderFiles[folder]; ok {
  342. rf.WithNeedTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  343. fs, de, by := sizeOfFile(f)
  344. nfiles += fs + de
  345. bytes += by
  346. return true
  347. })
  348. }
  349. bytes -= m.progressEmitter.BytesCompleted(folder)
  350. if debug {
  351. l.Debugf("%v NeedSize(%q): %d %d", m, folder, nfiles, bytes)
  352. }
  353. return
  354. }
  355. // NeedFolderFiles returns paginated list of currently needed files in
  356. // progress, queued, and to be queued on next puller iteration, as well as the
  357. // total number of files currently needed.
  358. func (m *Model) NeedFolderFiles(folder string, page, perpage int) ([]db.FileInfoTruncated, []db.FileInfoTruncated, []db.FileInfoTruncated, int) {
  359. m.fmut.RLock()
  360. defer m.fmut.RUnlock()
  361. total := 0
  362. rf, ok := m.folderFiles[folder]
  363. if !ok {
  364. return nil, nil, nil, 0
  365. }
  366. var progress, queued, rest []db.FileInfoTruncated
  367. var seen map[string]struct{}
  368. skip := (page - 1) * perpage
  369. get := perpage
  370. runner, ok := m.folderRunners[folder]
  371. if ok {
  372. allProgressNames, allQueuedNames := runner.Jobs()
  373. var progressNames, queuedNames []string
  374. progressNames, skip, get = getChunk(allProgressNames, skip, get)
  375. queuedNames, skip, get = getChunk(allQueuedNames, skip, get)
  376. progress = make([]db.FileInfoTruncated, len(progressNames))
  377. queued = make([]db.FileInfoTruncated, len(queuedNames))
  378. seen = make(map[string]struct{}, len(progressNames)+len(queuedNames))
  379. for i, name := range progressNames {
  380. if f, ok := rf.GetGlobalTruncated(name); ok {
  381. progress[i] = f
  382. seen[name] = struct{}{}
  383. }
  384. }
  385. for i, name := range queuedNames {
  386. if f, ok := rf.GetGlobalTruncated(name); ok {
  387. queued[i] = f
  388. seen[name] = struct{}{}
  389. }
  390. }
  391. }
  392. rest = make([]db.FileInfoTruncated, 0, perpage)
  393. rf.WithNeedTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  394. total++
  395. if skip > 0 {
  396. skip--
  397. return true
  398. }
  399. if get > 0 {
  400. ft := f.(db.FileInfoTruncated)
  401. if _, ok := seen[ft.Name]; !ok {
  402. rest = append(rest, ft)
  403. get--
  404. }
  405. }
  406. return true
  407. })
  408. return progress, queued, rest, total
  409. }
  410. // Index is called when a new device is connected and we receive their full index.
  411. // Implements the protocol.Model interface.
  412. func (m *Model) Index(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo, flags uint32, options []protocol.Option) {
  413. if flags != 0 {
  414. l.Warnln("protocol error: unknown flags 0x%x in Index message", flags)
  415. return
  416. }
  417. if debug {
  418. l.Debugf("IDX(in): %s %q: %d files", deviceID, folder, len(fs))
  419. }
  420. if !m.folderSharedWith(folder, deviceID) {
  421. events.Default.Log(events.FolderRejected, map[string]string{
  422. "folder": folder,
  423. "device": deviceID.String(),
  424. })
  425. l.Infof("Unexpected folder ID %q sent from device %q; ensure that the folder exists and that this device is selected under \"Share With\" in the folder configuration.", folder, deviceID)
  426. return
  427. }
  428. m.fmut.RLock()
  429. files, ok := m.folderFiles[folder]
  430. runner := m.folderRunners[folder]
  431. m.fmut.RUnlock()
  432. if runner != nil {
  433. // Runner may legitimately not be set if this is the "cleanup" Index
  434. // message at startup.
  435. defer runner.IndexUpdated()
  436. }
  437. if !ok {
  438. l.Fatalf("Index for nonexistant folder %q", folder)
  439. }
  440. for i := 0; i < len(fs); {
  441. if fs[i].Flags&^protocol.FlagsAll != 0 {
  442. if debug {
  443. l.Debugln("dropping update for file with unknown bits set", fs[i])
  444. }
  445. fs[i] = fs[len(fs)-1]
  446. fs = fs[:len(fs)-1]
  447. } else if symlinkInvalid(folder, fs[i]) {
  448. if debug {
  449. l.Debugln("dropping update for unsupported symlink", fs[i])
  450. }
  451. fs[i] = fs[len(fs)-1]
  452. fs = fs[:len(fs)-1]
  453. } else {
  454. i++
  455. }
  456. }
  457. files.Replace(deviceID, fs)
  458. events.Default.Log(events.RemoteIndexUpdated, map[string]interface{}{
  459. "device": deviceID.String(),
  460. "folder": folder,
  461. "items": len(fs),
  462. "version": files.LocalVersion(deviceID),
  463. })
  464. }
  465. // IndexUpdate is called for incremental updates to connected devices' indexes.
  466. // Implements the protocol.Model interface.
  467. func (m *Model) IndexUpdate(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo, flags uint32, options []protocol.Option) {
  468. if flags != 0 {
  469. l.Warnln("protocol error: unknown flags 0x%x in IndexUpdate message", flags)
  470. return
  471. }
  472. if debug {
  473. l.Debugf("%v IDXUP(in): %s / %q: %d files", m, deviceID, folder, len(fs))
  474. }
  475. if !m.folderSharedWith(folder, deviceID) {
  476. l.Infof("Update for unexpected folder ID %q sent from device %q; ensure that the folder exists and that this device is selected under \"Share With\" in the folder configuration.", folder, deviceID)
  477. return
  478. }
  479. m.fmut.RLock()
  480. files := m.folderFiles[folder]
  481. runner, ok := m.folderRunners[folder]
  482. m.fmut.RUnlock()
  483. if !ok {
  484. l.Fatalf("IndexUpdate for nonexistant folder %q", folder)
  485. }
  486. for i := 0; i < len(fs); {
  487. if fs[i].Flags&^protocol.FlagsAll != 0 {
  488. if debug {
  489. l.Debugln("dropping update for file with unknown bits set", fs[i])
  490. }
  491. fs[i] = fs[len(fs)-1]
  492. fs = fs[:len(fs)-1]
  493. } else if symlinkInvalid(folder, fs[i]) {
  494. if debug {
  495. l.Debugln("dropping update for unsupported symlink", fs[i])
  496. }
  497. fs[i] = fs[len(fs)-1]
  498. fs = fs[:len(fs)-1]
  499. } else {
  500. i++
  501. }
  502. }
  503. files.Update(deviceID, fs)
  504. events.Default.Log(events.RemoteIndexUpdated, map[string]interface{}{
  505. "device": deviceID.String(),
  506. "folder": folder,
  507. "items": len(fs),
  508. "version": files.LocalVersion(deviceID),
  509. })
  510. runner.IndexUpdated()
  511. }
  512. func (m *Model) folderSharedWith(folder string, deviceID protocol.DeviceID) bool {
  513. m.fmut.RLock()
  514. defer m.fmut.RUnlock()
  515. for _, nfolder := range m.deviceFolders[deviceID] {
  516. if nfolder == folder {
  517. return true
  518. }
  519. }
  520. return false
  521. }
  522. func (m *Model) ClusterConfig(deviceID protocol.DeviceID, cm protocol.ClusterConfigMessage) {
  523. m.pmut.Lock()
  524. if cm.ClientName == "syncthing" {
  525. m.deviceVer[deviceID] = cm.ClientVersion
  526. } else {
  527. m.deviceVer[deviceID] = cm.ClientName + " " + cm.ClientVersion
  528. }
  529. event := map[string]string{
  530. "id": deviceID.String(),
  531. "clientName": cm.ClientName,
  532. "clientVersion": cm.ClientVersion,
  533. }
  534. if conn, ok := m.rawConn[deviceID].(*tls.Conn); ok {
  535. event["addr"] = conn.RemoteAddr().String()
  536. }
  537. m.pmut.Unlock()
  538. events.Default.Log(events.DeviceConnected, event)
  539. l.Infof(`Device %s client is "%s %s"`, deviceID, cm.ClientName, cm.ClientVersion)
  540. var changed bool
  541. if name := cm.GetOption("name"); name != "" {
  542. l.Infof("Device %s name is %q", deviceID, name)
  543. device, ok := m.cfg.Devices()[deviceID]
  544. if ok && device.Name == "" {
  545. device.Name = name
  546. m.cfg.SetDevice(device)
  547. changed = true
  548. }
  549. }
  550. if m.cfg.Devices()[deviceID].Introducer {
  551. // This device is an introducer. Go through the announced lists of folders
  552. // and devices and add what we are missing.
  553. for _, folder := range cm.Folders {
  554. // If we don't have this folder yet, skip it. Ideally, we'd
  555. // offer up something in the GUI to create the folder, but for the
  556. // moment we only handle folders that we already have.
  557. if _, ok := m.folderDevices[folder.ID]; !ok {
  558. continue
  559. }
  560. nextDevice:
  561. for _, device := range folder.Devices {
  562. var id protocol.DeviceID
  563. copy(id[:], device.ID)
  564. if _, ok := m.cfg.Devices()[id]; !ok {
  565. // The device is currently unknown. Add it to the config.
  566. l.Infof("Adding device %v to config (vouched for by introducer %v)", id, deviceID)
  567. newDeviceCfg := config.DeviceConfiguration{
  568. DeviceID: id,
  569. Compression: m.cfg.Devices()[deviceID].Compression,
  570. Addresses: []string{"dynamic"},
  571. }
  572. // The introducers' introducers are also our introducers.
  573. if device.Flags&protocol.FlagIntroducer != 0 {
  574. l.Infof("Device %v is now also an introducer", id)
  575. newDeviceCfg.Introducer = true
  576. }
  577. m.cfg.SetDevice(newDeviceCfg)
  578. changed = true
  579. }
  580. for _, er := range m.deviceFolders[id] {
  581. if er == folder.ID {
  582. // We already share the folder with this device, so
  583. // nothing to do.
  584. continue nextDevice
  585. }
  586. }
  587. // We don't yet share this folder with this device. Add the device
  588. // to sharing list of the folder.
  589. l.Infof("Adding device %v to share %q (vouched for by introducer %v)", id, folder.ID, deviceID)
  590. m.deviceFolders[id] = append(m.deviceFolders[id], folder.ID)
  591. m.folderDevices[folder.ID] = append(m.folderDevices[folder.ID], id)
  592. folderCfg := m.cfg.Folders()[folder.ID]
  593. folderCfg.Devices = append(folderCfg.Devices, config.FolderDeviceConfiguration{
  594. DeviceID: id,
  595. })
  596. m.cfg.SetFolder(folderCfg)
  597. changed = true
  598. }
  599. }
  600. }
  601. if changed {
  602. m.cfg.Save()
  603. }
  604. }
  605. // Close removes the peer from the model and closes the underlying connection if possible.
  606. // Implements the protocol.Model interface.
  607. func (m *Model) Close(device protocol.DeviceID, err error) {
  608. l.Infof("Connection to %s closed: %v", device, err)
  609. events.Default.Log(events.DeviceDisconnected, map[string]string{
  610. "id": device.String(),
  611. "error": err.Error(),
  612. })
  613. m.pmut.Lock()
  614. m.fmut.RLock()
  615. for _, folder := range m.deviceFolders[device] {
  616. m.folderFiles[folder].Replace(device, nil)
  617. }
  618. m.fmut.RUnlock()
  619. conn, ok := m.rawConn[device]
  620. if ok {
  621. if conn, ok := conn.(*tls.Conn); ok {
  622. // If the underlying connection is a *tls.Conn, Close() does more
  623. // than it says on the tin. Specifically, it sends a TLS alert
  624. // message, which might block forever if the connection is dead
  625. // and we don't have a deadline site.
  626. conn.SetWriteDeadline(time.Now().Add(250 * time.Millisecond))
  627. }
  628. conn.Close()
  629. }
  630. delete(m.protoConn, device)
  631. delete(m.rawConn, device)
  632. delete(m.deviceVer, device)
  633. m.pmut.Unlock()
  634. }
  635. // Request returns the specified data segment by reading it from local disk.
  636. // Implements the protocol.Model interface.
  637. func (m *Model) Request(deviceID protocol.DeviceID, folder, name string, offset int64, size int, hash []byte, flags uint32, options []protocol.Option) ([]byte, error) {
  638. if offset < 0 || size < 0 {
  639. return nil, protocol.ErrNoSuchFile
  640. }
  641. if !m.folderSharedWith(folder, deviceID) {
  642. l.Warnf("Request from %s for file %s in unshared folder %q", deviceID, name, folder)
  643. return nil, protocol.ErrNoSuchFile
  644. }
  645. if flags != 0 {
  646. // We don't currently support or expect any flags.
  647. return nil, fmt.Errorf("protocol error: unknown flags 0x%x in Request message", flags)
  648. }
  649. // Verify that the requested file exists in the local model. We only need
  650. // to validate this file if we haven't done so recently, so we keep a
  651. // cache of successfull results. "Recently" can be quite a long time, as
  652. // we remove validation cache entries when we detect local changes. If
  653. // we're out of sync here and the file actually doesn't exist any more, or
  654. // has shrunk or something, then we'll anyway get a read error that we
  655. // pass on to the other side.
  656. m.rvmut.RLock()
  657. validated := m.reqValidationCache[folder+"/"+name]
  658. m.rvmut.RUnlock()
  659. if time.Since(validated) > reqValidationTime {
  660. m.fmut.RLock()
  661. folderFiles, ok := m.folderFiles[folder]
  662. m.fmut.RUnlock()
  663. if !ok {
  664. l.Warnf("Request from %s for file %s in nonexistent folder %q", deviceID, name, folder)
  665. return nil, protocol.ErrNoSuchFile
  666. }
  667. // This call is really expensive for large files, as we load the full
  668. // block list which may be megabytes and megabytes of data to allocate
  669. // space for, read, and deserialize.
  670. lf, ok := folderFiles.Get(protocol.LocalDeviceID, name)
  671. if !ok {
  672. return nil, protocol.ErrNoSuchFile
  673. }
  674. if lf.IsInvalid() || lf.IsDeleted() {
  675. if debug {
  676. l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d; invalid: %v", m, deviceID, folder, name, offset, size, lf)
  677. }
  678. return nil, protocol.ErrInvalid
  679. }
  680. if offset > lf.Size() {
  681. if debug {
  682. l.Debugf("%v REQ(in; nonexistent): %s: %q o=%d s=%d", m, deviceID, name, offset, size)
  683. }
  684. return nil, protocol.ErrNoSuchFile
  685. }
  686. m.rvmut.Lock()
  687. m.reqValidationCache[folder+"/"+name] = time.Now()
  688. if len(m.reqValidationCache) > reqValidationCacheSize {
  689. // Don't let the cache grow infinitely
  690. for name, validated := range m.reqValidationCache {
  691. if time.Since(validated) > time.Minute {
  692. delete(m.reqValidationCache, name)
  693. }
  694. }
  695. }
  696. m.rvmut.Unlock()
  697. }
  698. if debug && deviceID != protocol.LocalDeviceID {
  699. l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d", m, deviceID, folder, name, offset, size)
  700. }
  701. m.fmut.RLock()
  702. fn := filepath.Join(m.folderCfgs[folder].Path(), name)
  703. m.fmut.RUnlock()
  704. var reader io.ReaderAt
  705. var err error
  706. if info, err := os.Lstat(fn); err == nil && info.Mode()&os.ModeSymlink != 0 {
  707. target, _, err := symlinks.Read(fn)
  708. if err != nil {
  709. return nil, err
  710. }
  711. reader = strings.NewReader(target)
  712. } else {
  713. // Cannot easily cache fd's because we might need to delete the file
  714. // at any moment.
  715. reader, err = os.Open(fn)
  716. if err != nil {
  717. return nil, err
  718. }
  719. defer reader.(*os.File).Close()
  720. }
  721. buf := make([]byte, size)
  722. _, err = reader.ReadAt(buf, offset)
  723. if err != nil {
  724. return nil, err
  725. }
  726. return buf, nil
  727. }
  728. // ReplaceLocal replaces the local folder index with the given list of files.
  729. func (m *Model) ReplaceLocal(folder string, fs []protocol.FileInfo) {
  730. m.fmut.RLock()
  731. m.folderFiles[folder].ReplaceWithDelete(protocol.LocalDeviceID, fs, m.shortID)
  732. m.fmut.RUnlock()
  733. }
  734. func (m *Model) CurrentFolderFile(folder string, file string) (protocol.FileInfo, bool) {
  735. m.fmut.RLock()
  736. fs, ok := m.folderFiles[folder]
  737. m.fmut.RUnlock()
  738. if !ok {
  739. return protocol.FileInfo{}, false
  740. }
  741. f, ok := fs.Get(protocol.LocalDeviceID, file)
  742. return f, ok
  743. }
  744. func (m *Model) CurrentGlobalFile(folder string, file string) (protocol.FileInfo, bool) {
  745. m.fmut.RLock()
  746. fs, ok := m.folderFiles[folder]
  747. m.fmut.RUnlock()
  748. if !ok {
  749. return protocol.FileInfo{}, false
  750. }
  751. f, ok := fs.GetGlobal(file)
  752. return f, ok
  753. }
  754. type cFiler struct {
  755. m *Model
  756. r string
  757. }
  758. // Implements scanner.CurrentFiler
  759. func (cf cFiler) CurrentFile(file string) (protocol.FileInfo, bool) {
  760. return cf.m.CurrentFolderFile(cf.r, file)
  761. }
  762. // ConnectedTo returns true if we are connected to the named device.
  763. func (m *Model) ConnectedTo(deviceID protocol.DeviceID) bool {
  764. m.pmut.RLock()
  765. _, ok := m.protoConn[deviceID]
  766. m.pmut.RUnlock()
  767. if ok {
  768. m.deviceWasSeen(deviceID)
  769. }
  770. return ok
  771. }
  772. func (m *Model) GetIgnores(folder string) ([]string, []string, error) {
  773. var lines []string
  774. m.fmut.RLock()
  775. cfg, ok := m.folderCfgs[folder]
  776. m.fmut.RUnlock()
  777. if !ok {
  778. return lines, nil, fmt.Errorf("Folder %s does not exist", folder)
  779. }
  780. fd, err := os.Open(filepath.Join(cfg.Path(), ".stignore"))
  781. if err != nil {
  782. if os.IsNotExist(err) {
  783. return lines, nil, nil
  784. }
  785. l.Warnln("Loading .stignore:", err)
  786. return lines, nil, err
  787. }
  788. defer fd.Close()
  789. scanner := bufio.NewScanner(fd)
  790. for scanner.Scan() {
  791. lines = append(lines, strings.TrimSpace(scanner.Text()))
  792. }
  793. m.fmut.RLock()
  794. patterns := m.folderIgnores[folder].Patterns()
  795. m.fmut.RUnlock()
  796. return lines, patterns, nil
  797. }
  798. func (m *Model) SetIgnores(folder string, content []string) error {
  799. cfg, ok := m.folderCfgs[folder]
  800. if !ok {
  801. return fmt.Errorf("Folder %s does not exist", folder)
  802. }
  803. fd, err := ioutil.TempFile(cfg.Path(), ".syncthing.stignore-"+folder)
  804. if err != nil {
  805. l.Warnln("Saving .stignore:", err)
  806. return err
  807. }
  808. defer os.Remove(fd.Name())
  809. for _, line := range content {
  810. _, err = fmt.Fprintln(fd, line)
  811. if err != nil {
  812. l.Warnln("Saving .stignore:", err)
  813. return err
  814. }
  815. }
  816. err = fd.Close()
  817. if err != nil {
  818. l.Warnln("Saving .stignore:", err)
  819. return err
  820. }
  821. file := filepath.Join(cfg.Path(), ".stignore")
  822. err = osutil.Rename(fd.Name(), file)
  823. if err != nil {
  824. l.Warnln("Saving .stignore:", err)
  825. return err
  826. }
  827. return m.ScanFolder(folder)
  828. }
  829. // AddConnection adds a new peer connection to the model. An initial index will
  830. // be sent to the connected peer, thereafter index updates whenever the local
  831. // folder changes.
  832. func (m *Model) AddConnection(rawConn io.Closer, protoConn protocol.Connection) {
  833. deviceID := protoConn.ID()
  834. m.pmut.Lock()
  835. if _, ok := m.protoConn[deviceID]; ok {
  836. panic("add existing device")
  837. }
  838. m.protoConn[deviceID] = protoConn
  839. if _, ok := m.rawConn[deviceID]; ok {
  840. panic("add existing device")
  841. }
  842. m.rawConn[deviceID] = rawConn
  843. cm := m.clusterConfig(deviceID)
  844. protoConn.ClusterConfig(cm)
  845. m.fmut.RLock()
  846. for _, folder := range m.deviceFolders[deviceID] {
  847. fs := m.folderFiles[folder]
  848. go sendIndexes(protoConn, folder, fs, m.folderIgnores[folder])
  849. }
  850. m.fmut.RUnlock()
  851. m.pmut.Unlock()
  852. m.deviceWasSeen(deviceID)
  853. }
  854. func (m *Model) deviceStatRef(deviceID protocol.DeviceID) *stats.DeviceStatisticsReference {
  855. m.fmut.Lock()
  856. defer m.fmut.Unlock()
  857. if sr, ok := m.deviceStatRefs[deviceID]; ok {
  858. return sr
  859. }
  860. sr := stats.NewDeviceStatisticsReference(m.db, deviceID)
  861. m.deviceStatRefs[deviceID] = sr
  862. return sr
  863. }
  864. func (m *Model) deviceWasSeen(deviceID protocol.DeviceID) {
  865. m.deviceStatRef(deviceID).WasSeen()
  866. }
  867. func (m *Model) folderStatRef(folder string) *stats.FolderStatisticsReference {
  868. m.fmut.Lock()
  869. defer m.fmut.Unlock()
  870. sr, ok := m.folderStatRefs[folder]
  871. if !ok {
  872. sr = stats.NewFolderStatisticsReference(m.db, folder)
  873. m.folderStatRefs[folder] = sr
  874. }
  875. return sr
  876. }
  877. func (m *Model) receivedFile(folder string, file protocol.FileInfo) {
  878. m.folderStatRef(folder).ReceivedFile(file)
  879. }
  880. func sendIndexes(conn protocol.Connection, folder string, fs *db.FileSet, ignores *ignore.Matcher) {
  881. deviceID := conn.ID()
  882. name := conn.Name()
  883. var err error
  884. if debug {
  885. l.Debugf("sendIndexes for %s-%s/%q starting", deviceID, name, folder)
  886. }
  887. minLocalVer, err := sendIndexTo(true, 0, conn, folder, fs, ignores)
  888. for err == nil {
  889. time.Sleep(5 * time.Second)
  890. if fs.LocalVersion(protocol.LocalDeviceID) <= minLocalVer {
  891. continue
  892. }
  893. minLocalVer, err = sendIndexTo(false, minLocalVer, conn, folder, fs, ignores)
  894. }
  895. if debug {
  896. l.Debugf("sendIndexes for %s-%s/%q exiting: %v", deviceID, name, folder, err)
  897. }
  898. }
  899. func sendIndexTo(initial bool, minLocalVer int64, conn protocol.Connection, folder string, fs *db.FileSet, ignores *ignore.Matcher) (int64, error) {
  900. deviceID := conn.ID()
  901. name := conn.Name()
  902. batch := make([]protocol.FileInfo, 0, indexBatchSize)
  903. currentBatchSize := 0
  904. maxLocalVer := int64(0)
  905. var err error
  906. fs.WithHave(protocol.LocalDeviceID, func(fi db.FileIntf) bool {
  907. f := fi.(protocol.FileInfo)
  908. if f.LocalVersion <= minLocalVer {
  909. return true
  910. }
  911. if f.LocalVersion > maxLocalVer {
  912. maxLocalVer = f.LocalVersion
  913. }
  914. if ignores.Match(f.Name) || symlinkInvalid(folder, f) {
  915. if debug {
  916. l.Debugln("not sending update for ignored/unsupported symlink", f)
  917. }
  918. return true
  919. }
  920. if len(batch) == indexBatchSize || currentBatchSize > indexTargetSize {
  921. if initial {
  922. if err = conn.Index(folder, batch, 0, nil); err != nil {
  923. return false
  924. }
  925. if debug {
  926. l.Debugf("sendIndexes for %s-%s/%q: %d files (<%d bytes) (initial index)", deviceID, name, folder, len(batch), currentBatchSize)
  927. }
  928. initial = false
  929. } else {
  930. if err = conn.IndexUpdate(folder, batch, 0, nil); err != nil {
  931. return false
  932. }
  933. if debug {
  934. l.Debugf("sendIndexes for %s-%s/%q: %d files (<%d bytes) (batched update)", deviceID, name, folder, len(batch), currentBatchSize)
  935. }
  936. }
  937. batch = make([]protocol.FileInfo, 0, indexBatchSize)
  938. currentBatchSize = 0
  939. }
  940. batch = append(batch, f)
  941. currentBatchSize += indexPerFileSize + len(f.Blocks)*indexPerBlockSize
  942. return true
  943. })
  944. if initial && err == nil {
  945. err = conn.Index(folder, batch, 0, nil)
  946. if debug && err == nil {
  947. l.Debugf("sendIndexes for %s-%s/%q: %d files (small initial index)", deviceID, name, folder, len(batch))
  948. }
  949. } else if len(batch) > 0 && err == nil {
  950. err = conn.IndexUpdate(folder, batch, 0, nil)
  951. if debug && err == nil {
  952. l.Debugf("sendIndexes for %s-%s/%q: %d files (last batch)", deviceID, name, folder, len(batch))
  953. }
  954. }
  955. return maxLocalVer, err
  956. }
  957. func (m *Model) updateLocals(folder string, fs []protocol.FileInfo) {
  958. m.fmut.RLock()
  959. files := m.folderFiles[folder]
  960. m.fmut.RUnlock()
  961. files.Update(protocol.LocalDeviceID, fs)
  962. m.rvmut.Lock()
  963. for _, f := range fs {
  964. delete(m.reqValidationCache, folder+"/"+f.Name)
  965. }
  966. m.rvmut.Unlock()
  967. events.Default.Log(events.LocalIndexUpdated, map[string]interface{}{
  968. "folder": folder,
  969. "items": len(fs),
  970. "version": files.LocalVersion(protocol.LocalDeviceID),
  971. })
  972. }
  973. func (m *Model) requestGlobal(deviceID protocol.DeviceID, folder, name string, offset int64, size int, hash []byte, flags uint32, options []protocol.Option) ([]byte, error) {
  974. m.pmut.RLock()
  975. nc, ok := m.protoConn[deviceID]
  976. m.pmut.RUnlock()
  977. if !ok {
  978. return nil, fmt.Errorf("requestGlobal: no such device: %s", deviceID)
  979. }
  980. if debug {
  981. l.Debugf("%v REQ(out): %s: %q / %q o=%d s=%d h=%x f=%x op=%s", m, deviceID, folder, name, offset, size, hash, flags, options)
  982. }
  983. return nc.Request(folder, name, offset, size, hash, flags, options)
  984. }
  985. func (m *Model) AddFolder(cfg config.FolderConfiguration) {
  986. if m.started {
  987. panic("cannot add folder to started model")
  988. }
  989. if len(cfg.ID) == 0 {
  990. panic("cannot add empty folder id")
  991. }
  992. m.fmut.Lock()
  993. m.folderCfgs[cfg.ID] = cfg
  994. m.folderFiles[cfg.ID] = db.NewFileSet(cfg.ID, m.db)
  995. m.folderDevices[cfg.ID] = make([]protocol.DeviceID, len(cfg.Devices))
  996. for i, device := range cfg.Devices {
  997. m.folderDevices[cfg.ID][i] = device.DeviceID
  998. m.deviceFolders[device.DeviceID] = append(m.deviceFolders[device.DeviceID], cfg.ID)
  999. }
  1000. ignores := ignore.New(m.cfg.Options().CacheIgnoredFiles)
  1001. _ = ignores.Load(filepath.Join(cfg.Path(), ".stignore")) // Ignore error, there might not be an .stignore
  1002. m.folderIgnores[cfg.ID] = ignores
  1003. m.addedFolder = true
  1004. m.fmut.Unlock()
  1005. }
  1006. func (m *Model) ScanFolders() map[string]error {
  1007. m.fmut.RLock()
  1008. folders := make([]string, 0, len(m.folderCfgs))
  1009. for folder := range m.folderCfgs {
  1010. folders = append(folders, folder)
  1011. }
  1012. m.fmut.RUnlock()
  1013. errors := make(map[string]error, len(m.folderCfgs))
  1014. errorsMut := sync.NewMutex()
  1015. wg := sync.NewWaitGroup()
  1016. wg.Add(len(folders))
  1017. for _, folder := range folders {
  1018. folder := folder
  1019. go func() {
  1020. err := m.ScanFolder(folder)
  1021. if err != nil {
  1022. errorsMut.Lock()
  1023. errors[folder] = err
  1024. errorsMut.Unlock()
  1025. // Potentially sets the error twice, once in the scanner just
  1026. // by doing a check, and once here, if the error returned is
  1027. // the same one as returned by CheckFolderHealth, though
  1028. // duplicate set is handled by setError.
  1029. m.fmut.RLock()
  1030. srv := m.folderRunners[folder]
  1031. m.fmut.RUnlock()
  1032. srv.setError(err)
  1033. }
  1034. wg.Done()
  1035. }()
  1036. }
  1037. wg.Wait()
  1038. return errors
  1039. }
  1040. func (m *Model) ScanFolder(folder string) error {
  1041. return m.ScanFolderSubs(folder, nil)
  1042. }
  1043. func (m *Model) ScanFolderSubs(folder string, subs []string) error {
  1044. for i, sub := range subs {
  1045. sub = osutil.NativeFilename(sub)
  1046. if p := filepath.Clean(filepath.Join(folder, sub)); !strings.HasPrefix(p, folder) {
  1047. return errors.New("invalid subpath")
  1048. }
  1049. subs[i] = sub
  1050. }
  1051. m.fmut.Lock()
  1052. fs := m.folderFiles[folder]
  1053. folderCfg := m.folderCfgs[folder]
  1054. ignores := m.folderIgnores[folder]
  1055. runner, ok := m.folderRunners[folder]
  1056. m.fmut.Unlock()
  1057. // Folders are added to folderRunners only when they are started. We can't
  1058. // scan them before they have started, so that's what we need to check for
  1059. // here.
  1060. if !ok {
  1061. return errors.New("no such folder")
  1062. }
  1063. _ = ignores.Load(filepath.Join(folderCfg.Path(), ".stignore")) // Ignore error, there might not be an .stignore
  1064. // Required to make sure that we start indexing at a directory we're already
  1065. // aware off.
  1066. var unifySubs []string
  1067. nextSub:
  1068. for _, sub := range subs {
  1069. for sub != "" {
  1070. if _, ok = fs.Get(protocol.LocalDeviceID, sub); ok {
  1071. break
  1072. }
  1073. sub = filepath.Dir(sub)
  1074. if sub == "." || sub == string(filepath.Separator) {
  1075. sub = ""
  1076. }
  1077. }
  1078. for _, us := range unifySubs {
  1079. if strings.HasPrefix(sub, us) {
  1080. continue nextSub
  1081. }
  1082. }
  1083. unifySubs = append(unifySubs, sub)
  1084. }
  1085. subs = unifySubs
  1086. w := &scanner.Walker{
  1087. Dir: folderCfg.Path(),
  1088. Subs: subs,
  1089. Matcher: ignores,
  1090. BlockSize: protocol.BlockSize,
  1091. TempNamer: defTempNamer,
  1092. TempLifetime: time.Duration(m.cfg.Options().KeepTemporariesH) * time.Hour,
  1093. CurrentFiler: cFiler{m, folder},
  1094. MtimeRepo: db.NewVirtualMtimeRepo(m.db, folderCfg.ID),
  1095. IgnorePerms: folderCfg.IgnorePerms,
  1096. AutoNormalize: folderCfg.AutoNormalize,
  1097. Hashers: m.numHashers(folder),
  1098. ShortID: m.shortID,
  1099. }
  1100. runner.setState(FolderScanning)
  1101. fchan, err := w.Walk()
  1102. if err != nil {
  1103. // The error we get here is likely an OS level error, which might not be
  1104. // as readable as our health check errors. Check if we can get a health
  1105. // check error first, and use that if it's available.
  1106. if ferr := m.CheckFolderHealth(folder); ferr != nil {
  1107. err = ferr
  1108. }
  1109. runner.setError(err)
  1110. return err
  1111. }
  1112. batchSizeFiles := 100
  1113. batchSizeBlocks := 2048 // about 256 MB
  1114. batch := make([]protocol.FileInfo, 0, batchSizeFiles)
  1115. blocksHandled := 0
  1116. for f := range fchan {
  1117. if len(batch) == batchSizeFiles || blocksHandled > batchSizeBlocks {
  1118. if err := m.CheckFolderHealth(folder); err != nil {
  1119. l.Infof("Stopping folder %s mid-scan due to folder error: %s", folder, err)
  1120. return err
  1121. }
  1122. m.updateLocals(folder, batch)
  1123. batch = batch[:0]
  1124. blocksHandled = 0
  1125. }
  1126. batch = append(batch, f)
  1127. blocksHandled += len(f.Blocks)
  1128. }
  1129. if err := m.CheckFolderHealth(folder); err != nil {
  1130. l.Infof("Stopping folder %s mid-scan due to folder error: %s", folder, err)
  1131. return err
  1132. } else if len(batch) > 0 {
  1133. m.updateLocals(folder, batch)
  1134. }
  1135. batch = batch[:0]
  1136. // TODO: We should limit the Have scanning to start at sub
  1137. seenPrefix := false
  1138. var iterError error
  1139. fs.WithHaveTruncated(protocol.LocalDeviceID, func(fi db.FileIntf) bool {
  1140. f := fi.(db.FileInfoTruncated)
  1141. hasPrefix := len(subs) == 0
  1142. for _, sub := range subs {
  1143. if strings.HasPrefix(f.Name, sub) {
  1144. hasPrefix = true
  1145. break
  1146. }
  1147. }
  1148. // Return true so that we keep iterating, until we get to the part
  1149. // of the tree we are interested in. Then return false so we stop
  1150. // iterating when we've passed the end of the subtree.
  1151. if !hasPrefix {
  1152. return !seenPrefix
  1153. }
  1154. seenPrefix = true
  1155. if !f.IsDeleted() {
  1156. if f.IsInvalid() {
  1157. return true
  1158. }
  1159. if len(batch) == batchSizeFiles {
  1160. if err := m.CheckFolderHealth(folder); err != nil {
  1161. iterError = err
  1162. return false
  1163. }
  1164. m.updateLocals(folder, batch)
  1165. batch = batch[:0]
  1166. }
  1167. if ignores.Match(f.Name) || symlinkInvalid(folder, f) {
  1168. // File has been ignored or an unsupported symlink. Set invalid bit.
  1169. if debug {
  1170. l.Debugln("setting invalid bit on ignored", f)
  1171. }
  1172. nf := protocol.FileInfo{
  1173. Name: f.Name,
  1174. Flags: f.Flags | protocol.FlagInvalid,
  1175. Modified: f.Modified,
  1176. Version: f.Version, // The file is still the same, so don't bump version
  1177. }
  1178. batch = append(batch, nf)
  1179. } else if _, err := osutil.Lstat(filepath.Join(folderCfg.Path(), f.Name)); err != nil {
  1180. // File has been deleted.
  1181. // We don't specifically verify that the error is
  1182. // os.IsNotExist because there is a corner case when a
  1183. // directory is suddenly transformed into a file. When that
  1184. // happens, files that were in the directory (that is now a
  1185. // file) are deleted but will return a confusing error ("not a
  1186. // directory") when we try to Lstat() them.
  1187. nf := protocol.FileInfo{
  1188. Name: f.Name,
  1189. Flags: f.Flags | protocol.FlagDeleted,
  1190. Modified: f.Modified,
  1191. Version: f.Version.Update(m.shortID),
  1192. }
  1193. batch = append(batch, nf)
  1194. }
  1195. }
  1196. return true
  1197. })
  1198. if iterError != nil {
  1199. l.Infof("Stopping folder %s mid-scan due to folder error: %s", folder, iterError)
  1200. return iterError
  1201. }
  1202. if err := m.CheckFolderHealth(folder); err != nil {
  1203. l.Infof("Stopping folder %s mid-scan due to folder error: %s", folder, err)
  1204. return err
  1205. } else if len(batch) > 0 {
  1206. m.updateLocals(folder, batch)
  1207. }
  1208. runner.setState(FolderIdle)
  1209. return nil
  1210. }
  1211. func (m *Model) DelayScan(folder string, next time.Duration) {
  1212. m.fmut.Lock()
  1213. runner, ok := m.folderRunners[folder]
  1214. m.fmut.Unlock()
  1215. if !ok {
  1216. return
  1217. }
  1218. runner.DelayScan(next)
  1219. }
  1220. // numHashers returns the number of hasher routines to use for a given folder,
  1221. // taking into account configuration and available CPU cores.
  1222. func (m *Model) numHashers(folder string) int {
  1223. m.fmut.Lock()
  1224. folderCfg := m.folderCfgs[folder]
  1225. numFolders := len(m.folderCfgs)
  1226. m.fmut.Unlock()
  1227. if folderCfg.Hashers > 0 {
  1228. // Specific value set in the config, use that.
  1229. return folderCfg.Hashers
  1230. }
  1231. if perFolder := runtime.GOMAXPROCS(-1) / numFolders; perFolder > 0 {
  1232. // We have CPUs to spare, divide them per folder.
  1233. return perFolder
  1234. }
  1235. return 1
  1236. }
  1237. // clusterConfig returns a ClusterConfigMessage that is correct for the given peer device
  1238. func (m *Model) clusterConfig(device protocol.DeviceID) protocol.ClusterConfigMessage {
  1239. cm := protocol.ClusterConfigMessage{
  1240. ClientName: m.clientName,
  1241. ClientVersion: m.clientVersion,
  1242. Options: []protocol.Option{
  1243. {
  1244. Key: "name",
  1245. Value: m.deviceName,
  1246. },
  1247. },
  1248. }
  1249. m.fmut.RLock()
  1250. for _, folder := range m.deviceFolders[device] {
  1251. cr := protocol.Folder{
  1252. ID: folder,
  1253. }
  1254. for _, device := range m.folderDevices[folder] {
  1255. // DeviceID is a value type, but with an underlying array. Copy it
  1256. // so we don't grab aliases to the same array later on in device[:]
  1257. device := device
  1258. // TODO: Set read only bit when relevant
  1259. cn := protocol.Device{
  1260. ID: device[:],
  1261. Flags: protocol.FlagShareTrusted,
  1262. }
  1263. if deviceCfg := m.cfg.Devices()[device]; deviceCfg.Introducer {
  1264. cn.Flags |= protocol.FlagIntroducer
  1265. }
  1266. cr.Devices = append(cr.Devices, cn)
  1267. }
  1268. cm.Folders = append(cm.Folders, cr)
  1269. }
  1270. m.fmut.RUnlock()
  1271. return cm
  1272. }
  1273. func (m *Model) State(folder string) (string, time.Time, error) {
  1274. m.fmut.RLock()
  1275. runner, ok := m.folderRunners[folder]
  1276. m.fmut.RUnlock()
  1277. if !ok {
  1278. // The returned error should be an actual folder error, so returning
  1279. // errors.New("does not exist") or similar here would be
  1280. // inappropriate.
  1281. return "", time.Time{}, nil
  1282. }
  1283. state, changed, err := runner.getState()
  1284. return state.String(), changed, err
  1285. }
  1286. func (m *Model) Override(folder string) {
  1287. m.fmut.RLock()
  1288. fs, ok := m.folderFiles[folder]
  1289. runner := m.folderRunners[folder]
  1290. m.fmut.RUnlock()
  1291. if !ok {
  1292. return
  1293. }
  1294. runner.setState(FolderScanning)
  1295. batch := make([]protocol.FileInfo, 0, indexBatchSize)
  1296. fs.WithNeed(protocol.LocalDeviceID, func(fi db.FileIntf) bool {
  1297. need := fi.(protocol.FileInfo)
  1298. if len(batch) == indexBatchSize {
  1299. fs.Update(protocol.LocalDeviceID, batch)
  1300. batch = batch[:0]
  1301. }
  1302. have, ok := fs.Get(protocol.LocalDeviceID, need.Name)
  1303. if !ok || have.Name != need.Name {
  1304. // We are missing the file
  1305. need.Flags |= protocol.FlagDeleted
  1306. need.Blocks = nil
  1307. need.Version = need.Version.Update(m.shortID)
  1308. } else {
  1309. // We have the file, replace with our version
  1310. have.Version = have.Version.Merge(need.Version).Update(m.shortID)
  1311. need = have
  1312. }
  1313. need.LocalVersion = 0
  1314. batch = append(batch, need)
  1315. return true
  1316. })
  1317. if len(batch) > 0 {
  1318. fs.Update(protocol.LocalDeviceID, batch)
  1319. }
  1320. runner.setState(FolderIdle)
  1321. }
  1322. // CurrentLocalVersion returns the change version for the given folder.
  1323. // This is guaranteed to increment if the contents of the local folder has
  1324. // changed.
  1325. func (m *Model) CurrentLocalVersion(folder string) int64 {
  1326. m.fmut.RLock()
  1327. fs, ok := m.folderFiles[folder]
  1328. m.fmut.RUnlock()
  1329. if !ok {
  1330. // The folder might not exist, since this can be called with a user
  1331. // specified folder name from the REST interface.
  1332. return 0
  1333. }
  1334. return fs.LocalVersion(protocol.LocalDeviceID)
  1335. }
  1336. // RemoteLocalVersion returns the change version for the given folder, as
  1337. // sent by remote peers. This is guaranteed to increment if the contents of
  1338. // the remote or global folder has changed.
  1339. func (m *Model) RemoteLocalVersion(folder string) int64 {
  1340. m.fmut.RLock()
  1341. defer m.fmut.RUnlock()
  1342. fs, ok := m.folderFiles[folder]
  1343. if !ok {
  1344. // The folder might not exist, since this can be called with a user
  1345. // specified folder name from the REST interface.
  1346. return 0
  1347. }
  1348. var ver int64
  1349. for _, n := range m.folderDevices[folder] {
  1350. ver += fs.LocalVersion(n)
  1351. }
  1352. return ver
  1353. }
  1354. func (m *Model) GlobalDirectoryTree(folder, prefix string, levels int, dirsonly bool) map[string]interface{} {
  1355. m.fmut.RLock()
  1356. files, ok := m.folderFiles[folder]
  1357. m.fmut.RUnlock()
  1358. if !ok {
  1359. return nil
  1360. }
  1361. output := make(map[string]interface{})
  1362. sep := string(filepath.Separator)
  1363. prefix = osutil.NativeFilename(prefix)
  1364. if prefix != "" && !strings.HasSuffix(prefix, sep) {
  1365. prefix = prefix + sep
  1366. }
  1367. files.WithPrefixedGlobalTruncated(prefix, func(fi db.FileIntf) bool {
  1368. f := fi.(db.FileInfoTruncated)
  1369. if f.IsInvalid() || f.IsDeleted() || f.Name == prefix {
  1370. return true
  1371. }
  1372. f.Name = strings.Replace(f.Name, prefix, "", 1)
  1373. var dir, base string
  1374. if f.IsDirectory() && !f.IsSymlink() {
  1375. dir = f.Name
  1376. } else {
  1377. dir = filepath.Dir(f.Name)
  1378. base = filepath.Base(f.Name)
  1379. }
  1380. if levels > -1 && strings.Count(f.Name, sep) > levels {
  1381. return true
  1382. }
  1383. last := output
  1384. if dir != "." {
  1385. for _, path := range strings.Split(dir, sep) {
  1386. directory, ok := last[path]
  1387. if !ok {
  1388. newdir := make(map[string]interface{})
  1389. last[path] = newdir
  1390. last = newdir
  1391. } else {
  1392. last = directory.(map[string]interface{})
  1393. }
  1394. }
  1395. }
  1396. if !dirsonly && base != "" {
  1397. last[base] = []interface{}{
  1398. time.Unix(f.Modified, 0), f.Size(),
  1399. }
  1400. }
  1401. return true
  1402. })
  1403. return output
  1404. }
  1405. func (m *Model) Availability(folder, file string) []protocol.DeviceID {
  1406. // Acquire this lock first, as the value returned from foldersFiles can
  1407. // get heavily modified on Close()
  1408. m.pmut.RLock()
  1409. defer m.pmut.RUnlock()
  1410. m.fmut.RLock()
  1411. fs, ok := m.folderFiles[folder]
  1412. m.fmut.RUnlock()
  1413. if !ok {
  1414. return nil
  1415. }
  1416. availableDevices := []protocol.DeviceID{}
  1417. for _, device := range fs.Availability(file) {
  1418. _, ok := m.protoConn[device]
  1419. if ok {
  1420. availableDevices = append(availableDevices, device)
  1421. }
  1422. }
  1423. return availableDevices
  1424. }
  1425. // BringToFront bumps the given files priority in the job queue.
  1426. func (m *Model) BringToFront(folder, file string) {
  1427. m.pmut.RLock()
  1428. defer m.pmut.RUnlock()
  1429. runner, ok := m.folderRunners[folder]
  1430. if ok {
  1431. runner.BringToFront(file)
  1432. }
  1433. }
  1434. // CheckFolderHealth checks the folder for common errors and returns the
  1435. // current folder error, or nil if the folder is healthy.
  1436. func (m *Model) CheckFolderHealth(id string) error {
  1437. folder, ok := m.cfg.Folders()[id]
  1438. if !ok {
  1439. return errors.New("folder does not exist")
  1440. }
  1441. fi, err := os.Stat(folder.Path())
  1442. if m.CurrentLocalVersion(id) > 0 {
  1443. // Safety check. If the cached index contains files but the
  1444. // folder doesn't exist, we have a problem. We would assume
  1445. // that all files have been deleted which might not be the case,
  1446. // so mark it as invalid instead.
  1447. if err != nil || !fi.IsDir() {
  1448. err = errors.New("folder path missing")
  1449. } else if !folder.HasMarker() {
  1450. err = errors.New("folder marker missing")
  1451. }
  1452. } else if os.IsNotExist(err) {
  1453. // If we don't have any files in the index, and the directory
  1454. // doesn't exist, try creating it.
  1455. err = osutil.MkdirAll(folder.Path(), 0700)
  1456. if err == nil {
  1457. err = folder.CreateMarker()
  1458. }
  1459. } else if !folder.HasMarker() {
  1460. // If we don't have any files in the index, and the path does exist
  1461. // but the marker is not there, create it.
  1462. err = folder.CreateMarker()
  1463. }
  1464. m.fmut.RLock()
  1465. runner, runnerExists := m.folderRunners[folder.ID]
  1466. m.fmut.RUnlock()
  1467. var oldErr error
  1468. if runnerExists {
  1469. _, _, oldErr = runner.getState()
  1470. }
  1471. if err != nil {
  1472. if oldErr != nil && oldErr.Error() != err.Error() {
  1473. l.Infof("Folder %q error changed: %q -> %q", folder.ID, oldErr, err)
  1474. } else if oldErr == nil {
  1475. l.Warnf("Stopping folder %q - %v", folder.ID, err)
  1476. }
  1477. if runnerExists {
  1478. runner.setError(err)
  1479. }
  1480. } else if oldErr != nil {
  1481. l.Infof("Folder %q error is cleared, restarting", folder.ID)
  1482. if runnerExists {
  1483. runner.clearError()
  1484. }
  1485. }
  1486. return err
  1487. }
  1488. func (m *Model) ResetFolder(folder string) error {
  1489. for _, f := range db.ListFolders(m.db) {
  1490. if f == folder {
  1491. l.Infof("Cleaning data for folder %q", folder)
  1492. db.DropFolder(m.db, folder)
  1493. return nil
  1494. }
  1495. }
  1496. return fmt.Errorf("Unknown folder %q", folder)
  1497. }
  1498. func (m *Model) String() string {
  1499. return fmt.Sprintf("model@%p", m)
  1500. }
  1501. func (m *Model) VerifyConfiguration(from, to config.Configuration) error {
  1502. return nil
  1503. }
  1504. func (m *Model) CommitConfiguration(from, to config.Configuration) bool {
  1505. // TODO: This should not use reflect, and should take more care to try to handle stuff without restart.
  1506. // Adding, removing or changing folders requires restart
  1507. if !reflect.DeepEqual(from.Folders, to.Folders) {
  1508. return false
  1509. }
  1510. // Removing a device requres restart
  1511. toDevs := make(map[protocol.DeviceID]bool, len(from.Devices))
  1512. for _, dev := range to.Devices {
  1513. toDevs[dev.DeviceID] = true
  1514. }
  1515. for _, dev := range from.Devices {
  1516. if _, ok := toDevs[dev.DeviceID]; !ok {
  1517. return false
  1518. }
  1519. }
  1520. // All of the generic options require restart
  1521. if !reflect.DeepEqual(from.Options, to.Options) {
  1522. return false
  1523. }
  1524. return true
  1525. }
  1526. func symlinkInvalid(folder string, fi db.FileIntf) bool {
  1527. if !symlinks.Supported && fi.IsSymlink() && !fi.IsInvalid() && !fi.IsDeleted() {
  1528. symlinkWarning.Do(func() {
  1529. l.Warnln("Symlinks are disabled, unsupported or require Administrator privileges. This might cause your folder to appear out of sync.")
  1530. })
  1531. // Need to type switch for the concrete type to be able to access fields...
  1532. var name string
  1533. switch fi := fi.(type) {
  1534. case protocol.FileInfo:
  1535. name = fi.Name
  1536. case db.FileInfoTruncated:
  1537. name = fi.Name
  1538. }
  1539. l.Infoln("Unsupported symlink", name, "in folder", folder)
  1540. return true
  1541. }
  1542. return false
  1543. }
  1544. // Skips `skip` elements and retrieves up to `get` elements from a given slice.
  1545. // Returns the resulting slice, plus how much elements are left to skip or
  1546. // copy to satisfy the values which were provided, given the slice is not
  1547. // big enough.
  1548. func getChunk(data []string, skip, get int) ([]string, int, int) {
  1549. l := len(data)
  1550. if l <= skip {
  1551. return []string{}, skip - l, get
  1552. } else if l < skip+get {
  1553. return data[skip:l], 0, get - (l - skip)
  1554. }
  1555. return data[skip : skip+get], 0, 0
  1556. }