model.go 84 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877
  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 https://mozilla.org/MPL/2.0/.
  6. package model
  7. import (
  8. "bytes"
  9. "context"
  10. "encoding/json"
  11. "fmt"
  12. "net"
  13. "path/filepath"
  14. "reflect"
  15. "regexp"
  16. "runtime"
  17. "strings"
  18. stdsync "sync"
  19. "time"
  20. "github.com/pkg/errors"
  21. "github.com/thejerf/suture"
  22. "github.com/syncthing/syncthing/lib/config"
  23. "github.com/syncthing/syncthing/lib/connections"
  24. "github.com/syncthing/syncthing/lib/db"
  25. "github.com/syncthing/syncthing/lib/events"
  26. "github.com/syncthing/syncthing/lib/fs"
  27. "github.com/syncthing/syncthing/lib/ignore"
  28. "github.com/syncthing/syncthing/lib/osutil"
  29. "github.com/syncthing/syncthing/lib/protocol"
  30. "github.com/syncthing/syncthing/lib/scanner"
  31. "github.com/syncthing/syncthing/lib/stats"
  32. "github.com/syncthing/syncthing/lib/sync"
  33. "github.com/syncthing/syncthing/lib/upgrade"
  34. "github.com/syncthing/syncthing/lib/util"
  35. "github.com/syncthing/syncthing/lib/versioner"
  36. )
  37. // How many files to send in each Index/IndexUpdate message.
  38. const (
  39. maxBatchSizeBytes = 250 * 1024 // Aim for making index messages no larger than 250 KiB (uncompressed)
  40. maxBatchSizeFiles = 1000 // Either way, don't include more files than this
  41. )
  42. type service interface {
  43. BringToFront(string)
  44. Override()
  45. Revert()
  46. DelayScan(d time.Duration)
  47. SchedulePull() // something relevant changed, we should try a pull
  48. Jobs(page, perpage int) ([]string, []string, int) // In progress, Queued, skipped
  49. Scan(subs []string) error
  50. Serve()
  51. Stop()
  52. CheckHealth() error
  53. Errors() []FileError
  54. WatchError() error
  55. ForceRescan(file protocol.FileInfo) error
  56. GetStatistics() (stats.FolderStatistics, error)
  57. getState() (folderState, time.Time, error)
  58. }
  59. type Availability struct {
  60. ID protocol.DeviceID `json:"id"`
  61. FromTemporary bool `json:"fromTemporary"`
  62. }
  63. type Model interface {
  64. suture.Service
  65. connections.Model
  66. ResetFolder(folder string)
  67. DelayScan(folder string, next time.Duration)
  68. ScanFolder(folder string) error
  69. ScanFolders() map[string]error
  70. ScanFolderSubdirs(folder string, subs []string) error
  71. State(folder string) (string, time.Time, error)
  72. FolderErrors(folder string) ([]FileError, error)
  73. WatchError(folder string) error
  74. Override(folder string)
  75. Revert(folder string)
  76. BringToFront(folder, file string)
  77. GetIgnores(folder string) ([]string, []string, error)
  78. SetIgnores(folder string, content []string) error
  79. GetFolderVersions(folder string) (map[string][]versioner.FileVersion, error)
  80. RestoreFolderVersions(folder string, versions map[string]time.Time) (map[string]string, error)
  81. LocalChangedFiles(folder string, page, perpage int) []db.FileInfoTruncated
  82. NeedFolderFiles(folder string, page, perpage int) ([]db.FileInfoTruncated, []db.FileInfoTruncated, []db.FileInfoTruncated)
  83. RemoteNeedFolderFiles(device protocol.DeviceID, folder string, page, perpage int) ([]db.FileInfoTruncated, error)
  84. CurrentFolderFile(folder string, file string) (protocol.FileInfo, bool)
  85. CurrentGlobalFile(folder string, file string) (protocol.FileInfo, bool)
  86. Availability(folder string, file protocol.FileInfo, block protocol.BlockInfo) []Availability
  87. GlobalSize(folder string) db.Counts
  88. LocalSize(folder string) db.Counts
  89. NeedSize(folder string) db.Counts
  90. ReceiveOnlyChangedSize(folder string) db.Counts
  91. CurrentSequence(folder string) (int64, bool)
  92. RemoteSequence(folder string) (int64, bool)
  93. Completion(device protocol.DeviceID, folder string) FolderCompletion
  94. ConnectionStats() map[string]interface{}
  95. DeviceStatistics() (map[string]stats.DeviceStatistics, error)
  96. FolderStatistics() (map[string]stats.FolderStatistics, error)
  97. UsageReportingStats(version int, preview bool) map[string]interface{}
  98. StartDeadlockDetector(timeout time.Duration)
  99. GlobalDirectoryTree(folder, prefix string, levels int, dirsonly bool) map[string]interface{}
  100. }
  101. type model struct {
  102. *suture.Supervisor
  103. cfg config.Wrapper
  104. db *db.Lowlevel
  105. finder *db.BlockFinder
  106. progressEmitter *ProgressEmitter
  107. id protocol.DeviceID
  108. shortID protocol.ShortID
  109. cacheIgnoredFiles bool
  110. protectedFiles []string
  111. evLogger events.Logger
  112. clientName string
  113. clientVersion string
  114. fmut sync.RWMutex // protects the below
  115. folderCfgs map[string]config.FolderConfiguration // folder -> cfg
  116. folderFiles map[string]*db.FileSet // folder -> files
  117. deviceStatRefs map[protocol.DeviceID]*stats.DeviceStatisticsReference // deviceID -> statsRef
  118. folderIgnores map[string]*ignore.Matcher // folder -> matcher object
  119. folderRunners map[string]service // folder -> puller or scanner
  120. folderRunnerTokens map[string][]suture.ServiceToken // folder -> tokens for puller or scanner
  121. folderRestartMuts syncMutexMap // folder -> restart mutex
  122. folderVersioners map[string]versioner.Versioner // folder -> versioner (may be nil)
  123. pmut sync.RWMutex // protects the below
  124. conn map[protocol.DeviceID]connections.Connection
  125. connRequestLimiters map[protocol.DeviceID]*byteSemaphore
  126. closed map[protocol.DeviceID]chan struct{}
  127. helloMessages map[protocol.DeviceID]protocol.HelloResult
  128. deviceDownloads map[protocol.DeviceID]*deviceDownloadState
  129. remotePausedFolders map[protocol.DeviceID][]string // deviceID -> folders
  130. foldersRunning int32 // for testing only
  131. }
  132. type folderFactory func(*model, *db.FileSet, *ignore.Matcher, config.FolderConfiguration, versioner.Versioner, fs.Filesystem, events.Logger) service
  133. var (
  134. folderFactories = make(map[config.FolderType]folderFactory)
  135. )
  136. var (
  137. errDeviceUnknown = errors.New("unknown device")
  138. errDevicePaused = errors.New("device is paused")
  139. errDeviceIgnored = errors.New("device is ignored")
  140. ErrFolderPaused = errors.New("folder is paused")
  141. errFolderNotRunning = errors.New("folder is not running")
  142. errFolderMissing = errors.New("no such folder")
  143. errNetworkNotAllowed = errors.New("network not allowed")
  144. errNoVersioner = errors.New("folder has no versioner")
  145. // errors about why a connection is closed
  146. errIgnoredFolderRemoved = errors.New("folder no longer ignored")
  147. errReplacingConnection = errors.New("replacing connection")
  148. errStopped = errors.New("Syncthing is being stopped")
  149. )
  150. // NewModel creates and starts a new model. The model starts in read-only mode,
  151. // where it sends index information to connected peers and responds to requests
  152. // for file data without altering the local folder in any way.
  153. func NewModel(cfg config.Wrapper, id protocol.DeviceID, clientName, clientVersion string, ldb *db.Lowlevel, protectedFiles []string, evLogger events.Logger) Model {
  154. m := &model{
  155. Supervisor: suture.New("model", suture.Spec{
  156. Log: func(line string) {
  157. l.Debugln(line)
  158. },
  159. PassThroughPanics: true,
  160. }),
  161. cfg: cfg,
  162. db: ldb,
  163. finder: db.NewBlockFinder(ldb),
  164. progressEmitter: NewProgressEmitter(cfg, evLogger),
  165. id: id,
  166. shortID: id.Short(),
  167. cacheIgnoredFiles: cfg.Options().CacheIgnoredFiles,
  168. protectedFiles: protectedFiles,
  169. evLogger: evLogger,
  170. clientName: clientName,
  171. clientVersion: clientVersion,
  172. folderCfgs: make(map[string]config.FolderConfiguration),
  173. folderFiles: make(map[string]*db.FileSet),
  174. deviceStatRefs: make(map[protocol.DeviceID]*stats.DeviceStatisticsReference),
  175. folderIgnores: make(map[string]*ignore.Matcher),
  176. folderRunners: make(map[string]service),
  177. folderRunnerTokens: make(map[string][]suture.ServiceToken),
  178. folderVersioners: make(map[string]versioner.Versioner),
  179. conn: make(map[protocol.DeviceID]connections.Connection),
  180. connRequestLimiters: make(map[protocol.DeviceID]*byteSemaphore),
  181. closed: make(map[protocol.DeviceID]chan struct{}),
  182. helloMessages: make(map[protocol.DeviceID]protocol.HelloResult),
  183. deviceDownloads: make(map[protocol.DeviceID]*deviceDownloadState),
  184. remotePausedFolders: make(map[protocol.DeviceID][]string),
  185. fmut: sync.NewRWMutex(),
  186. pmut: sync.NewRWMutex(),
  187. }
  188. for devID := range cfg.Devices() {
  189. m.deviceStatRefs[devID] = stats.NewDeviceStatisticsReference(m.db, devID.String())
  190. }
  191. m.Add(m.progressEmitter)
  192. scanLimiter.setCapacity(cfg.Options().MaxConcurrentScans)
  193. return m
  194. }
  195. func (m *model) Serve() {
  196. m.onServe()
  197. m.Supervisor.Serve()
  198. }
  199. func (m *model) ServeBackground() {
  200. m.onServe()
  201. m.Supervisor.ServeBackground()
  202. }
  203. func (m *model) onServe() {
  204. // Add and start folders
  205. for _, folderCfg := range m.cfg.Folders() {
  206. if folderCfg.Paused {
  207. folderCfg.CreateRoot()
  208. continue
  209. }
  210. m.newFolder(folderCfg)
  211. }
  212. m.cfg.Subscribe(m)
  213. }
  214. func (m *model) Stop() {
  215. m.cfg.Unsubscribe(m)
  216. m.Supervisor.Stop()
  217. devs := m.cfg.Devices()
  218. ids := make([]protocol.DeviceID, 0, len(devs))
  219. for id := range devs {
  220. ids = append(ids, id)
  221. }
  222. w := m.closeConns(ids, errStopped)
  223. w.Wait()
  224. }
  225. // StartDeadlockDetector starts a deadlock detector on the models locks which
  226. // causes panics in case the locks cannot be acquired in the given timeout
  227. // period.
  228. func (m *model) StartDeadlockDetector(timeout time.Duration) {
  229. l.Infof("Starting deadlock detector with %v timeout", timeout)
  230. detector := newDeadlockDetector(timeout)
  231. detector.Watch("fmut", m.fmut)
  232. detector.Watch("pmut", m.pmut)
  233. }
  234. // startFolder constructs the folder service and starts it.
  235. func (m *model) startFolder(folder string) {
  236. m.fmut.RLock()
  237. folderCfg := m.folderCfgs[folder]
  238. m.fmut.RUnlock()
  239. // Close connections to affected devices
  240. m.closeConns(folderCfg.DeviceIDs(), fmt.Errorf("started folder %v", folderCfg.Description()))
  241. m.fmut.Lock()
  242. defer m.fmut.Unlock()
  243. m.startFolderLocked(folderCfg)
  244. }
  245. // Need to hold lock on m.fmut when calling this.
  246. func (m *model) startFolderLocked(cfg config.FolderConfiguration) {
  247. _, ok := m.folderRunners[cfg.ID]
  248. if ok {
  249. l.Warnln("Cannot start already running folder", cfg.Description())
  250. panic("cannot start already running folder")
  251. }
  252. folderFactory, ok := folderFactories[cfg.Type]
  253. if !ok {
  254. panic(fmt.Sprintf("unknown folder type 0x%x", cfg.Type))
  255. }
  256. folder := cfg.ID
  257. fset := m.folderFiles[folder]
  258. // Find any devices for which we hold the index in the db, but the folder
  259. // is not shared, and drop it.
  260. expected := mapDevices(cfg.DeviceIDs())
  261. for _, available := range fset.ListDevices() {
  262. if _, ok := expected[available]; !ok {
  263. l.Debugln("dropping", folder, "state for", available)
  264. fset.Drop(available)
  265. }
  266. }
  267. v, ok := fset.Sequence(protocol.LocalDeviceID), true
  268. indexHasFiles := ok && v > 0
  269. if !indexHasFiles {
  270. // It's a blank folder, so this may the first time we're looking at
  271. // it. Attempt to create and tag with our marker as appropriate. We
  272. // don't really do anything with errors at this point except warn -
  273. // if these things don't work, we still want to start the folder and
  274. // it'll show up as errored later.
  275. if err := cfg.CreateRoot(); err != nil {
  276. l.Warnln("Failed to create folder root directory", err)
  277. } else if err = cfg.CreateMarker(); err != nil {
  278. l.Warnln("Failed to create folder marker:", err)
  279. }
  280. }
  281. ffs := fset.MtimeFS()
  282. // These are our metadata files, and they should always be hidden.
  283. _ = ffs.Hide(config.DefaultMarkerName)
  284. _ = ffs.Hide(".stversions")
  285. _ = ffs.Hide(".stignore")
  286. var ver versioner.Versioner
  287. if cfg.Versioning.Type != "" {
  288. var err error
  289. ver, err = versioner.New(ffs, cfg.Versioning)
  290. if err != nil {
  291. panic(fmt.Errorf("creating versioner: %v", err))
  292. }
  293. if service, ok := ver.(suture.Service); ok {
  294. // The versioner implements the suture.Service interface, so
  295. // expects to be run in the background in addition to being called
  296. // when files are going to be archived.
  297. token := m.Add(service)
  298. m.folderRunnerTokens[folder] = append(m.folderRunnerTokens[folder], token)
  299. }
  300. }
  301. m.folderVersioners[folder] = ver
  302. ignores := m.folderIgnores[folder]
  303. p := folderFactory(m, fset, ignores, cfg, ver, ffs, m.evLogger)
  304. m.folderRunners[folder] = p
  305. m.warnAboutOverwritingProtectedFiles(cfg, ignores)
  306. token := m.Add(p)
  307. m.folderRunnerTokens[folder] = append(m.folderRunnerTokens[folder], token)
  308. l.Infof("Ready to synchronize %s (%s)", cfg.Description(), cfg.Type)
  309. }
  310. func (m *model) warnAboutOverwritingProtectedFiles(cfg config.FolderConfiguration, ignores *ignore.Matcher) {
  311. if cfg.Type == config.FolderTypeSendOnly {
  312. return
  313. }
  314. // This is a bit of a hack.
  315. ffs := cfg.Filesystem()
  316. if ffs.Type() != fs.FilesystemTypeBasic {
  317. return
  318. }
  319. folderLocation := ffs.URI()
  320. var filesAtRisk []string
  321. for _, protectedFilePath := range m.protectedFiles {
  322. // check if file is synced in this folder
  323. if protectedFilePath != folderLocation && !fs.IsParent(protectedFilePath, folderLocation) {
  324. continue
  325. }
  326. // check if file is ignored
  327. relPath, _ := filepath.Rel(folderLocation, protectedFilePath)
  328. if ignores.Match(relPath).IsIgnored() {
  329. continue
  330. }
  331. filesAtRisk = append(filesAtRisk, protectedFilePath)
  332. }
  333. if len(filesAtRisk) > 0 {
  334. l.Warnln("Some protected files may be overwritten and cause issues. See https://docs.syncthing.net/users/config.html#syncing-configuration-files for more information. The at risk files are:", strings.Join(filesAtRisk, ", "))
  335. }
  336. }
  337. func (m *model) addFolder(cfg config.FolderConfiguration) {
  338. if len(cfg.ID) == 0 {
  339. panic("cannot add empty folder id")
  340. }
  341. if len(cfg.Path) == 0 {
  342. panic("cannot add empty folder path")
  343. }
  344. // Creating the fileset can take a long time (metadata calculation) so
  345. // we do it outside of the lock.
  346. fset := db.NewFileSet(cfg.ID, cfg.Filesystem(), m.db)
  347. m.fmut.Lock()
  348. defer m.fmut.Unlock()
  349. m.addFolderLocked(cfg, fset)
  350. }
  351. func (m *model) addFolderLocked(cfg config.FolderConfiguration, fset *db.FileSet) {
  352. m.folderCfgs[cfg.ID] = cfg
  353. m.folderFiles[cfg.ID] = fset
  354. ignores := ignore.New(cfg.Filesystem(), ignore.WithCache(m.cacheIgnoredFiles))
  355. if err := ignores.Load(".stignore"); err != nil && !fs.IsNotExist(err) {
  356. l.Warnln("Loading ignores:", err)
  357. }
  358. m.folderIgnores[cfg.ID] = ignores
  359. }
  360. func (m *model) removeFolder(cfg config.FolderConfiguration) {
  361. m.stopFolder(cfg, fmt.Errorf("removing folder %v", cfg.Description()))
  362. m.fmut.Lock()
  363. isPathUnique := true
  364. for folderID, folderCfg := range m.folderCfgs {
  365. if folderID != cfg.ID && folderCfg.Path == cfg.Path {
  366. isPathUnique = false
  367. break
  368. }
  369. }
  370. if isPathUnique {
  371. // Delete syncthing specific files
  372. cfg.Filesystem().RemoveAll(config.DefaultMarkerName)
  373. }
  374. m.removeFolderLocked(cfg)
  375. m.fmut.Unlock()
  376. // Remove it from the database
  377. db.DropFolder(m.db, cfg.ID)
  378. }
  379. func (m *model) stopFolder(cfg config.FolderConfiguration, err error) {
  380. // Stop the services running for this folder and wait for them to finish
  381. // stopping to prevent races on restart.
  382. m.fmut.RLock()
  383. tokens := m.folderRunnerTokens[cfg.ID]
  384. m.fmut.RUnlock()
  385. for _, id := range tokens {
  386. m.RemoveAndWait(id, 0)
  387. }
  388. // Wait for connections to stop to ensure that no more calls to methods
  389. // expecting this folder to exist happen (e.g. .IndexUpdate).
  390. m.closeConns(cfg.DeviceIDs(), err).Wait()
  391. }
  392. // Need to hold lock on m.fmut when calling this.
  393. func (m *model) removeFolderLocked(cfg config.FolderConfiguration) {
  394. // Clean up our config maps
  395. delete(m.folderCfgs, cfg.ID)
  396. delete(m.folderFiles, cfg.ID)
  397. delete(m.folderIgnores, cfg.ID)
  398. delete(m.folderRunners, cfg.ID)
  399. delete(m.folderRunnerTokens, cfg.ID)
  400. delete(m.folderVersioners, cfg.ID)
  401. }
  402. func (m *model) restartFolder(from, to config.FolderConfiguration) {
  403. if len(to.ID) == 0 {
  404. panic("bug: cannot restart empty folder ID")
  405. }
  406. if to.ID != from.ID {
  407. l.Warnf("bug: folder restart cannot change ID %q -> %q", from.ID, to.ID)
  408. panic("bug: folder restart cannot change ID")
  409. }
  410. // This mutex protects the entirety of the restart operation, preventing
  411. // there from being more than one folder restart operation in progress
  412. // at any given time. The usual fmut/pmut stuff doesn't cover this,
  413. // because those locks are released while we are waiting for the folder
  414. // to shut down (and must be so because the folder might need them as
  415. // part of its operations before shutting down).
  416. restartMut := m.folderRestartMuts.Get(to.ID)
  417. restartMut.Lock()
  418. defer restartMut.Unlock()
  419. var infoMsg string
  420. var errMsg string
  421. switch {
  422. case to.Paused:
  423. infoMsg = "Paused"
  424. errMsg = "pausing"
  425. case from.Paused:
  426. infoMsg = "Unpaused"
  427. errMsg = "unpausing"
  428. default:
  429. infoMsg = "Restarted"
  430. errMsg = "restarting"
  431. }
  432. var fset *db.FileSet
  433. if !to.Paused {
  434. // Creating the fileset can take a long time (metadata calculation)
  435. // so we do it outside of the lock.
  436. fset = db.NewFileSet(to.ID, to.Filesystem(), m.db)
  437. }
  438. m.stopFolder(from, fmt.Errorf("%v folder %v", errMsg, to.Description()))
  439. m.fmut.Lock()
  440. defer m.fmut.Unlock()
  441. m.removeFolderLocked(from)
  442. if !to.Paused {
  443. m.addFolderLocked(to, fset)
  444. m.startFolderLocked(to)
  445. }
  446. l.Infof("%v folder %v (%v)", infoMsg, to.Description(), to.Type)
  447. }
  448. func (m *model) newFolder(cfg config.FolderConfiguration) {
  449. // Creating the fileset can take a long time (metadata calculation) so
  450. // we do it outside of the lock.
  451. fset := db.NewFileSet(cfg.ID, cfg.Filesystem(), m.db)
  452. // Close connections to affected devices
  453. m.closeConns(cfg.DeviceIDs(), fmt.Errorf("started folder %v", cfg.Description()))
  454. m.fmut.Lock()
  455. defer m.fmut.Unlock()
  456. m.addFolderLocked(cfg, fset)
  457. m.startFolderLocked(cfg)
  458. }
  459. func (m *model) UsageReportingStats(version int, preview bool) map[string]interface{} {
  460. stats := make(map[string]interface{})
  461. if version >= 3 {
  462. // Block stats
  463. blockStatsMut.Lock()
  464. copyBlockStats := make(map[string]int)
  465. for k, v := range blockStats {
  466. copyBlockStats[k] = v
  467. if !preview {
  468. blockStats[k] = 0
  469. }
  470. }
  471. blockStatsMut.Unlock()
  472. stats["blockStats"] = copyBlockStats
  473. // Transport stats
  474. m.pmut.RLock()
  475. transportStats := make(map[string]int)
  476. for _, conn := range m.conn {
  477. transportStats[conn.Transport()]++
  478. }
  479. m.pmut.RUnlock()
  480. stats["transportStats"] = transportStats
  481. // Ignore stats
  482. ignoreStats := map[string]int{
  483. "lines": 0,
  484. "inverts": 0,
  485. "folded": 0,
  486. "deletable": 0,
  487. "rooted": 0,
  488. "includes": 0,
  489. "escapedIncludes": 0,
  490. "doubleStars": 0,
  491. "stars": 0,
  492. }
  493. var seenPrefix [3]bool
  494. for folder := range m.cfg.Folders() {
  495. lines, _, err := m.GetIgnores(folder)
  496. if err != nil {
  497. continue
  498. }
  499. ignoreStats["lines"] += len(lines)
  500. for _, line := range lines {
  501. // Allow prefixes to be specified in any order, but only once.
  502. for {
  503. if strings.HasPrefix(line, "!") && !seenPrefix[0] {
  504. seenPrefix[0] = true
  505. line = line[1:]
  506. ignoreStats["inverts"] += 1
  507. } else if strings.HasPrefix(line, "(?i)") && !seenPrefix[1] {
  508. seenPrefix[1] = true
  509. line = line[4:]
  510. ignoreStats["folded"] += 1
  511. } else if strings.HasPrefix(line, "(?d)") && !seenPrefix[2] {
  512. seenPrefix[2] = true
  513. line = line[4:]
  514. ignoreStats["deletable"] += 1
  515. } else {
  516. seenPrefix[0] = false
  517. seenPrefix[1] = false
  518. seenPrefix[2] = false
  519. break
  520. }
  521. }
  522. // Noops, remove
  523. line = strings.TrimSuffix(line, "**")
  524. line = strings.TrimPrefix(line, "**/")
  525. if strings.HasPrefix(line, "/") {
  526. ignoreStats["rooted"] += 1
  527. } else if strings.HasPrefix(line, "#include ") {
  528. ignoreStats["includes"] += 1
  529. if strings.Contains(line, "..") {
  530. ignoreStats["escapedIncludes"] += 1
  531. }
  532. }
  533. if strings.Contains(line, "**") {
  534. ignoreStats["doubleStars"] += 1
  535. // Remove not to trip up star checks.
  536. line = strings.Replace(line, "**", "", -1)
  537. }
  538. if strings.Contains(line, "*") {
  539. ignoreStats["stars"] += 1
  540. }
  541. }
  542. }
  543. stats["ignoreStats"] = ignoreStats
  544. }
  545. return stats
  546. }
  547. type ConnectionInfo struct {
  548. protocol.Statistics
  549. Connected bool
  550. Paused bool
  551. Address string
  552. ClientVersion string
  553. Type string
  554. Crypto string
  555. }
  556. func (info ConnectionInfo) MarshalJSON() ([]byte, error) {
  557. return json.Marshal(map[string]interface{}{
  558. "at": info.At,
  559. "inBytesTotal": info.InBytesTotal,
  560. "outBytesTotal": info.OutBytesTotal,
  561. "connected": info.Connected,
  562. "paused": info.Paused,
  563. "address": info.Address,
  564. "clientVersion": info.ClientVersion,
  565. "type": info.Type,
  566. "crypto": info.Crypto,
  567. })
  568. }
  569. // ConnectionStats returns a map with connection statistics for each device.
  570. func (m *model) ConnectionStats() map[string]interface{} {
  571. m.pmut.RLock()
  572. defer m.pmut.RUnlock()
  573. res := make(map[string]interface{})
  574. devs := m.cfg.Devices()
  575. conns := make(map[string]ConnectionInfo, len(devs))
  576. for device, deviceCfg := range devs {
  577. hello := m.helloMessages[device]
  578. versionString := hello.ClientVersion
  579. if hello.ClientName != "syncthing" {
  580. versionString = hello.ClientName + " " + hello.ClientVersion
  581. }
  582. ci := ConnectionInfo{
  583. ClientVersion: strings.TrimSpace(versionString),
  584. Paused: deviceCfg.Paused,
  585. }
  586. if conn, ok := m.conn[device]; ok {
  587. ci.Type = conn.Type()
  588. ci.Crypto = conn.Crypto()
  589. ci.Connected = ok
  590. ci.Statistics = conn.Statistics()
  591. if addr := conn.RemoteAddr(); addr != nil {
  592. ci.Address = addr.String()
  593. }
  594. }
  595. conns[device.String()] = ci
  596. }
  597. res["connections"] = conns
  598. in, out := protocol.TotalInOut()
  599. res["total"] = ConnectionInfo{
  600. Statistics: protocol.Statistics{
  601. At: time.Now(),
  602. InBytesTotal: in,
  603. OutBytesTotal: out,
  604. },
  605. }
  606. return res
  607. }
  608. // DeviceStatistics returns statistics about each device
  609. func (m *model) DeviceStatistics() (map[string]stats.DeviceStatistics, error) {
  610. m.fmut.RLock()
  611. defer m.fmut.RUnlock()
  612. res := make(map[string]stats.DeviceStatistics, len(m.deviceStatRefs))
  613. for id, sr := range m.deviceStatRefs {
  614. stats, err := sr.GetStatistics()
  615. if err != nil {
  616. return nil, err
  617. }
  618. res[id.String()] = stats
  619. }
  620. return res, nil
  621. }
  622. // FolderStatistics returns statistics about each folder
  623. func (m *model) FolderStatistics() (map[string]stats.FolderStatistics, error) {
  624. res := make(map[string]stats.FolderStatistics)
  625. m.fmut.RLock()
  626. defer m.fmut.RUnlock()
  627. for id, runner := range m.folderRunners {
  628. stats, err := runner.GetStatistics()
  629. if err != nil {
  630. return nil, err
  631. }
  632. res[id] = stats
  633. }
  634. return res, nil
  635. }
  636. type FolderCompletion struct {
  637. CompletionPct float64
  638. NeedBytes int64
  639. NeedItems int64
  640. GlobalBytes int64
  641. NeedDeletes int64
  642. }
  643. // Map returns the members as a map, e.g. used in api to serialize as Json.
  644. func (comp FolderCompletion) Map() map[string]interface{} {
  645. return map[string]interface{}{
  646. "completion": comp.CompletionPct,
  647. "needBytes": comp.NeedBytes,
  648. "needItems": comp.NeedItems,
  649. "globalBytes": comp.GlobalBytes,
  650. "needDeletes": comp.NeedDeletes,
  651. }
  652. }
  653. // Completion returns the completion status, in percent, for the given device
  654. // and folder.
  655. func (m *model) Completion(device protocol.DeviceID, folder string) FolderCompletion {
  656. m.fmut.RLock()
  657. rf, ok := m.folderFiles[folder]
  658. m.fmut.RUnlock()
  659. if !ok {
  660. return FolderCompletion{} // Folder doesn't exist, so we hardly have any of it
  661. }
  662. tot := rf.GlobalSize().Bytes
  663. if tot == 0 {
  664. // Folder is empty, so we have all of it
  665. return FolderCompletion{
  666. CompletionPct: 100,
  667. }
  668. }
  669. m.pmut.RLock()
  670. counts := m.deviceDownloads[device].GetBlockCounts(folder)
  671. m.pmut.RUnlock()
  672. var need, items, fileNeed, downloaded, deletes int64
  673. rf.WithNeedTruncated(device, func(f db.FileIntf) bool {
  674. ft := f.(db.FileInfoTruncated)
  675. // If the file is deleted, we account it only in the deleted column.
  676. if ft.Deleted {
  677. deletes++
  678. return true
  679. }
  680. // This might might be more than it really is, because some blocks can be of a smaller size.
  681. downloaded = int64(counts[ft.Name]) * int64(ft.BlockSize())
  682. fileNeed = ft.FileSize() - downloaded
  683. if fileNeed < 0 {
  684. fileNeed = 0
  685. }
  686. need += fileNeed
  687. items++
  688. return true
  689. })
  690. needRatio := float64(need) / float64(tot)
  691. completionPct := 100 * (1 - needRatio)
  692. // If the completion is 100% but there are deletes we need to handle,
  693. // drop it down a notch. Hack for consumers that look only at the
  694. // percentage (our own GUI does the same calculation as here on its own
  695. // and needs the same fixup).
  696. if need == 0 && deletes > 0 {
  697. completionPct = 95 // chosen by fair dice roll
  698. }
  699. l.Debugf("%v Completion(%s, %q): %f (%d / %d = %f)", m, device, folder, completionPct, need, tot, needRatio)
  700. return FolderCompletion{
  701. CompletionPct: completionPct,
  702. NeedBytes: need,
  703. NeedItems: items,
  704. GlobalBytes: tot,
  705. NeedDeletes: deletes,
  706. }
  707. }
  708. func addSizeOfFile(s *db.Counts, f db.FileIntf) {
  709. switch {
  710. case f.IsDeleted():
  711. s.Deleted++
  712. case f.IsDirectory():
  713. s.Directories++
  714. case f.IsSymlink():
  715. s.Symlinks++
  716. default:
  717. s.Files++
  718. }
  719. s.Bytes += f.FileSize()
  720. }
  721. // GlobalSize returns the number of files, deleted files and total bytes for all
  722. // files in the global model.
  723. func (m *model) GlobalSize(folder string) db.Counts {
  724. m.fmut.RLock()
  725. rf, ok := m.folderFiles[folder]
  726. m.fmut.RUnlock()
  727. if ok {
  728. return rf.GlobalSize()
  729. }
  730. return db.Counts{}
  731. }
  732. // LocalSize returns the number of files, deleted files and total bytes for all
  733. // files in the local folder.
  734. func (m *model) LocalSize(folder string) db.Counts {
  735. m.fmut.RLock()
  736. rf, ok := m.folderFiles[folder]
  737. m.fmut.RUnlock()
  738. if ok {
  739. return rf.LocalSize()
  740. }
  741. return db.Counts{}
  742. }
  743. // ReceiveOnlyChangedSize returns the number of files, deleted files and
  744. // total bytes for all files that have changed locally in a receieve only
  745. // folder.
  746. func (m *model) ReceiveOnlyChangedSize(folder string) db.Counts {
  747. m.fmut.RLock()
  748. rf, ok := m.folderFiles[folder]
  749. m.fmut.RUnlock()
  750. if ok {
  751. return rf.ReceiveOnlyChangedSize()
  752. }
  753. return db.Counts{}
  754. }
  755. // NeedSize returns the number of currently needed files and their total size
  756. // minus the amount that has already been downloaded.
  757. func (m *model) NeedSize(folder string) db.Counts {
  758. m.fmut.RLock()
  759. rf, ok := m.folderFiles[folder]
  760. cfg := m.folderCfgs[folder]
  761. m.fmut.RUnlock()
  762. var result db.Counts
  763. if ok {
  764. rf.WithNeedTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  765. if cfg.IgnoreDelete && f.IsDeleted() {
  766. return true
  767. }
  768. addSizeOfFile(&result, f)
  769. return true
  770. })
  771. }
  772. result.Bytes -= m.progressEmitter.BytesCompleted(folder)
  773. // This may happen if we are in progress of pulling files that were
  774. // deleted globally after the pull started.
  775. if result.Bytes < 0 {
  776. result.Bytes = 0
  777. }
  778. l.Debugf("%v NeedSize(%q): %v", m, folder, result)
  779. return result
  780. }
  781. // NeedFolderFiles returns paginated list of currently needed files in
  782. // progress, queued, and to be queued on next puller iteration.
  783. func (m *model) NeedFolderFiles(folder string, page, perpage int) ([]db.FileInfoTruncated, []db.FileInfoTruncated, []db.FileInfoTruncated) {
  784. m.fmut.RLock()
  785. rf, rfOk := m.folderFiles[folder]
  786. runner, runnerOk := m.folderRunners[folder]
  787. cfg := m.folderCfgs[folder]
  788. m.fmut.RUnlock()
  789. if !rfOk {
  790. return nil, nil, nil
  791. }
  792. var progress, queued, rest []db.FileInfoTruncated
  793. var seen map[string]struct{}
  794. skip := (page - 1) * perpage
  795. get := perpage
  796. if runnerOk {
  797. progressNames, queuedNames, skipped := runner.Jobs(page, perpage)
  798. progress = make([]db.FileInfoTruncated, len(progressNames))
  799. queued = make([]db.FileInfoTruncated, len(queuedNames))
  800. seen = make(map[string]struct{}, len(progressNames)+len(queuedNames))
  801. for i, name := range progressNames {
  802. if f, ok := rf.GetGlobalTruncated(name); ok {
  803. progress[i] = f
  804. seen[name] = struct{}{}
  805. }
  806. }
  807. for i, name := range queuedNames {
  808. if f, ok := rf.GetGlobalTruncated(name); ok {
  809. queued[i] = f
  810. seen[name] = struct{}{}
  811. }
  812. }
  813. get -= len(seen)
  814. if get == 0 {
  815. return progress, queued, nil
  816. }
  817. skip -= skipped
  818. }
  819. rest = make([]db.FileInfoTruncated, 0, perpage)
  820. rf.WithNeedTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  821. if cfg.IgnoreDelete && f.IsDeleted() {
  822. return true
  823. }
  824. if skip > 0 {
  825. skip--
  826. return true
  827. }
  828. ft := f.(db.FileInfoTruncated)
  829. if _, ok := seen[ft.Name]; !ok {
  830. rest = append(rest, ft)
  831. get--
  832. }
  833. return get > 0
  834. })
  835. return progress, queued, rest
  836. }
  837. // LocalChangedFiles returns a paginated list of currently needed files in
  838. // progress, queued, and to be queued on next puller iteration, as well as the
  839. // total number of files currently needed.
  840. func (m *model) LocalChangedFiles(folder string, page, perpage int) []db.FileInfoTruncated {
  841. m.fmut.RLock()
  842. rf, ok := m.folderFiles[folder]
  843. fcfg := m.folderCfgs[folder]
  844. m.fmut.RUnlock()
  845. if !ok {
  846. return nil
  847. }
  848. if fcfg.Type != config.FolderTypeReceiveOnly {
  849. return nil
  850. }
  851. if rf.ReceiveOnlyChangedSize().TotalItems() == 0 {
  852. return nil
  853. }
  854. files := make([]db.FileInfoTruncated, 0, perpage)
  855. skip := (page - 1) * perpage
  856. get := perpage
  857. rf.WithHaveTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  858. if !f.IsReceiveOnlyChanged() {
  859. return true
  860. }
  861. if skip > 0 {
  862. skip--
  863. return true
  864. }
  865. ft := f.(db.FileInfoTruncated)
  866. files = append(files, ft)
  867. get--
  868. return get > 0
  869. })
  870. return files
  871. }
  872. // RemoteNeedFolderFiles returns paginated list of currently needed files in
  873. // progress, queued, and to be queued on next puller iteration, as well as the
  874. // total number of files currently needed.
  875. func (m *model) RemoteNeedFolderFiles(device protocol.DeviceID, folder string, page, perpage int) ([]db.FileInfoTruncated, error) {
  876. m.fmut.RLock()
  877. m.pmut.RLock()
  878. err := m.checkDeviceFolderConnectedLocked(device, folder)
  879. rf := m.folderFiles[folder]
  880. m.pmut.RUnlock()
  881. m.fmut.RUnlock()
  882. if err != nil {
  883. return nil, err
  884. }
  885. files := make([]db.FileInfoTruncated, 0, perpage)
  886. skip := (page - 1) * perpage
  887. get := perpage
  888. rf.WithNeedTruncated(device, func(f db.FileIntf) bool {
  889. if skip > 0 {
  890. skip--
  891. return true
  892. }
  893. files = append(files, f.(db.FileInfoTruncated))
  894. get--
  895. return get > 0
  896. })
  897. return files, nil
  898. }
  899. // Index is called when a new device is connected and we receive their full index.
  900. // Implements the protocol.Model interface.
  901. func (m *model) Index(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo) error {
  902. return m.handleIndex(deviceID, folder, fs, false)
  903. }
  904. // IndexUpdate is called for incremental updates to connected devices' indexes.
  905. // Implements the protocol.Model interface.
  906. func (m *model) IndexUpdate(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo) error {
  907. return m.handleIndex(deviceID, folder, fs, true)
  908. }
  909. func (m *model) handleIndex(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo, update bool) error {
  910. op := "Index"
  911. if update {
  912. op += " update"
  913. }
  914. l.Debugf("%v (in): %s / %q: %d files", op, deviceID, folder, len(fs))
  915. if cfg, ok := m.cfg.Folder(folder); !ok || !cfg.SharedWith(deviceID) {
  916. l.Infof("%v 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.", op, folder, deviceID)
  917. return errors.Wrap(errFolderMissing, folder)
  918. } else if cfg.Paused {
  919. l.Debugf("%v for paused folder (ID %q) sent from device %q.", op, folder, deviceID)
  920. return errors.Wrap(ErrFolderPaused, folder)
  921. }
  922. m.fmut.RLock()
  923. files, existing := m.folderFiles[folder]
  924. runner, running := m.folderRunners[folder]
  925. m.fmut.RUnlock()
  926. if !existing {
  927. l.Infof("%v for nonexistent folder %q", op, folder)
  928. return errors.Wrap(errFolderMissing, folder)
  929. }
  930. if running {
  931. defer runner.SchedulePull()
  932. }
  933. m.pmut.RLock()
  934. downloads := m.deviceDownloads[deviceID]
  935. m.pmut.RUnlock()
  936. downloads.Update(folder, makeForgetUpdate(fs))
  937. if !update {
  938. files.Drop(deviceID)
  939. }
  940. for i := range fs {
  941. // The local flags should never be transmitted over the wire. Make
  942. // sure they look like they weren't.
  943. fs[i].LocalFlags = 0
  944. }
  945. files.Update(deviceID, fs)
  946. m.evLogger.Log(events.RemoteIndexUpdated, map[string]interface{}{
  947. "device": deviceID.String(),
  948. "folder": folder,
  949. "items": len(fs),
  950. "version": files.Sequence(deviceID),
  951. })
  952. return nil
  953. }
  954. func (m *model) ClusterConfig(deviceID protocol.DeviceID, cm protocol.ClusterConfig) error {
  955. // Check the peer device's announced folders against our own. Emits events
  956. // for folders that we don't expect (unknown or not shared).
  957. // Also, collect a list of folders we do share, and if he's interested in
  958. // temporary indexes, subscribe the connection.
  959. tempIndexFolders := make([]string, 0, len(cm.Folders))
  960. m.pmut.RLock()
  961. conn, ok := m.conn[deviceID]
  962. closed := m.closed[deviceID]
  963. hello := m.helloMessages[deviceID]
  964. m.pmut.RUnlock()
  965. if !ok {
  966. panic("bug: ClusterConfig called on closed or nonexistent connection")
  967. }
  968. changed := false
  969. deviceCfg := m.cfg.Devices()[deviceID]
  970. // See issue #3802 - in short, we can't send modern symlink entries to older
  971. // clients.
  972. dropSymlinks := false
  973. if hello.ClientName == m.clientName && upgrade.CompareVersions(hello.ClientVersion, "v0.14.14") < 0 {
  974. l.Warnln("Not sending symlinks to old client", deviceID, "- please upgrade to v0.14.14 or newer")
  975. dropSymlinks = true
  976. }
  977. // Needs to happen outside of the fmut, as can cause CommitConfiguration
  978. if deviceCfg.AutoAcceptFolders {
  979. for _, folder := range cm.Folders {
  980. changed = m.handleAutoAccepts(deviceCfg, folder) || changed
  981. }
  982. }
  983. m.fmut.RLock()
  984. var paused []string
  985. for _, folder := range cm.Folders {
  986. cfg, ok := m.cfg.Folder(folder.ID)
  987. if !ok || !cfg.SharedWith(deviceID) {
  988. if deviceCfg.IgnoredFolder(folder.ID) {
  989. l.Infof("Ignoring folder %s from device %s since we are configured to", folder.Description(), deviceID)
  990. continue
  991. }
  992. m.cfg.AddOrUpdatePendingFolder(folder.ID, folder.Label, deviceID)
  993. changed = true
  994. m.evLogger.Log(events.FolderRejected, map[string]string{
  995. "folder": folder.ID,
  996. "folderLabel": folder.Label,
  997. "device": deviceID.String(),
  998. })
  999. l.Infof("Unexpected folder %s sent from device %q; ensure that the folder exists and that this device is selected under \"Share With\" in the folder configuration.", folder.Description(), deviceID)
  1000. continue
  1001. }
  1002. if folder.Paused {
  1003. paused = append(paused, folder.ID)
  1004. continue
  1005. }
  1006. if cfg.Paused {
  1007. continue
  1008. }
  1009. fs, ok := m.folderFiles[folder.ID]
  1010. if !ok {
  1011. // Shouldn't happen because !cfg.Paused, but might happen
  1012. // if the folder is about to be unpaused, but not yet.
  1013. continue
  1014. }
  1015. if !folder.DisableTempIndexes {
  1016. tempIndexFolders = append(tempIndexFolders, folder.ID)
  1017. }
  1018. myIndexID := fs.IndexID(protocol.LocalDeviceID)
  1019. mySequence := fs.Sequence(protocol.LocalDeviceID)
  1020. var startSequence int64
  1021. for _, dev := range folder.Devices {
  1022. if dev.ID == m.id {
  1023. // This is the other side's description of what it knows
  1024. // about us. Lets check to see if we can start sending index
  1025. // updates directly or need to send the index from start...
  1026. if dev.IndexID == myIndexID {
  1027. // They say they've seen our index ID before, so we can
  1028. // send a delta update only.
  1029. if dev.MaxSequence > mySequence {
  1030. // Safety check. They claim to have more or newer
  1031. // index data than we have - either we have lost
  1032. // index data, or reset the index without resetting
  1033. // the IndexID, or something else weird has
  1034. // happened. We send a full index to reset the
  1035. // situation.
  1036. l.Infof("Device %v folder %s is delta index compatible, but seems out of sync with reality", deviceID, folder.Description())
  1037. startSequence = 0
  1038. continue
  1039. }
  1040. l.Debugf("Device %v folder %s is delta index compatible (mlv=%d)", deviceID, folder.Description(), dev.MaxSequence)
  1041. startSequence = dev.MaxSequence
  1042. } else if dev.IndexID != 0 {
  1043. // They say they've seen an index ID from us, but it's
  1044. // not the right one. Either they are confused or we
  1045. // must have reset our database since last talking to
  1046. // them. We'll start with a full index transfer.
  1047. l.Infof("Device %v folder %s has mismatching index ID for us (%v != %v)", deviceID, folder.Description(), dev.IndexID, myIndexID)
  1048. startSequence = 0
  1049. }
  1050. } else if dev.ID == deviceID {
  1051. // This is the other side's description of themselves. We
  1052. // check to see that it matches the IndexID we have on file,
  1053. // otherwise we drop our old index data and expect to get a
  1054. // completely new set.
  1055. theirIndexID := fs.IndexID(deviceID)
  1056. if dev.IndexID == 0 {
  1057. // They're not announcing an index ID. This means they
  1058. // do not support delta indexes and we should clear any
  1059. // information we have from them before accepting their
  1060. // index, which will presumably be a full index.
  1061. fs.Drop(deviceID)
  1062. } else if dev.IndexID != theirIndexID {
  1063. // The index ID we have on file is not what they're
  1064. // announcing. They must have reset their database and
  1065. // will probably send us a full index. We drop any
  1066. // information we have and remember this new index ID
  1067. // instead.
  1068. l.Infof("Device %v folder %s has a new index ID (%v)", deviceID, folder.Description(), dev.IndexID)
  1069. fs.Drop(deviceID)
  1070. fs.SetIndexID(deviceID, dev.IndexID)
  1071. } else {
  1072. // They're sending a recognized index ID and will most
  1073. // likely use delta indexes. We might already have files
  1074. // that we need to pull so let the folder runner know
  1075. // that it should recheck the index data.
  1076. if runner := m.folderRunners[folder.ID]; runner != nil {
  1077. defer runner.SchedulePull()
  1078. }
  1079. }
  1080. }
  1081. }
  1082. is := &indexSender{
  1083. conn: conn,
  1084. connClosed: closed,
  1085. folder: folder.ID,
  1086. fset: fs,
  1087. prevSequence: startSequence,
  1088. dropSymlinks: dropSymlinks,
  1089. evLogger: m.evLogger,
  1090. }
  1091. is.Service = util.AsService(is.serve, is.String())
  1092. // The token isn't tracked as the service stops when the connection
  1093. // terminates and is automatically removed from supervisor (by
  1094. // implementing suture.IsCompletable).
  1095. m.Add(is)
  1096. }
  1097. m.fmut.RUnlock()
  1098. m.pmut.Lock()
  1099. m.remotePausedFolders[deviceID] = paused
  1100. m.pmut.Unlock()
  1101. // This breaks if we send multiple CM messages during the same connection.
  1102. if len(tempIndexFolders) > 0 {
  1103. m.pmut.RLock()
  1104. conn, ok := m.conn[deviceID]
  1105. m.pmut.RUnlock()
  1106. // In case we've got ClusterConfig, and the connection disappeared
  1107. // from infront of our nose.
  1108. if ok {
  1109. m.progressEmitter.temporaryIndexSubscribe(conn, tempIndexFolders)
  1110. }
  1111. }
  1112. if deviceCfg.Introducer {
  1113. folders, devices, foldersDevices, introduced := m.handleIntroductions(deviceCfg, cm)
  1114. folders, devices, deintroduced := m.handleDeintroductions(deviceCfg, foldersDevices, folders, devices)
  1115. if introduced || deintroduced {
  1116. changed = true
  1117. cfg := m.cfg.RawCopy()
  1118. cfg.Folders = make([]config.FolderConfiguration, 0, len(folders))
  1119. for _, fcfg := range folders {
  1120. cfg.Folders = append(cfg.Folders, fcfg)
  1121. }
  1122. cfg.Devices = make([]config.DeviceConfiguration, len(devices))
  1123. for _, dcfg := range devices {
  1124. cfg.Devices = append(cfg.Devices, dcfg)
  1125. }
  1126. m.cfg.Replace(cfg)
  1127. }
  1128. }
  1129. if changed {
  1130. if err := m.cfg.Save(); err != nil {
  1131. l.Warnln("Failed to save config", err)
  1132. }
  1133. }
  1134. return nil
  1135. }
  1136. // handleIntroductions handles adding devices/folders that are shared by an introducer device
  1137. func (m *model) handleIntroductions(introducerCfg config.DeviceConfiguration, cm protocol.ClusterConfig) (map[string]config.FolderConfiguration, map[protocol.DeviceID]config.DeviceConfiguration, folderDeviceSet, bool) {
  1138. changed := false
  1139. folders := m.cfg.Folders()
  1140. devices := m.cfg.Devices()
  1141. foldersDevices := make(folderDeviceSet)
  1142. for _, folder := range cm.Folders {
  1143. // Adds devices which we do not have, but the introducer has
  1144. // for the folders that we have in common. Also, shares folders
  1145. // with devices that we have in common, yet are currently not sharing
  1146. // the folder.
  1147. fcfg, ok := folders[folder.ID]
  1148. if !ok {
  1149. // Don't have this folder, carry on.
  1150. continue
  1151. }
  1152. folderChanged := false
  1153. for _, device := range folder.Devices {
  1154. // No need to share with self.
  1155. if device.ID == m.id {
  1156. continue
  1157. }
  1158. foldersDevices.set(device.ID, folder.ID)
  1159. if _, ok := m.cfg.Devices()[device.ID]; !ok {
  1160. // The device is currently unknown. Add it to the config.
  1161. devices[device.ID] = m.introduceDevice(device, introducerCfg)
  1162. } else if fcfg.SharedWith(device.ID) {
  1163. // We already share the folder with this device, so
  1164. // nothing to do.
  1165. continue
  1166. }
  1167. // We don't yet share this folder with this device. Add the device
  1168. // to sharing list of the folder.
  1169. l.Infof("Sharing folder %s with %v (vouched for by introducer %v)", folder.Description(), device.ID, introducerCfg.DeviceID)
  1170. fcfg.Devices = append(fcfg.Devices, config.FolderDeviceConfiguration{
  1171. DeviceID: device.ID,
  1172. IntroducedBy: introducerCfg.DeviceID,
  1173. })
  1174. folderChanged = true
  1175. }
  1176. if folderChanged {
  1177. folders[fcfg.ID] = fcfg
  1178. changed = true
  1179. }
  1180. }
  1181. return folders, devices, foldersDevices, changed
  1182. }
  1183. // handleDeintroductions handles removals of devices/shares that are removed by an introducer device
  1184. func (m *model) handleDeintroductions(introducerCfg config.DeviceConfiguration, foldersDevices folderDeviceSet, folders map[string]config.FolderConfiguration, devices map[protocol.DeviceID]config.DeviceConfiguration) (map[string]config.FolderConfiguration, map[protocol.DeviceID]config.DeviceConfiguration, bool) {
  1185. if introducerCfg.SkipIntroductionRemovals {
  1186. return folders, devices, false
  1187. }
  1188. changed := false
  1189. devicesNotIntroduced := make(map[protocol.DeviceID]struct{})
  1190. // Check if we should unshare some folders, if the introducer has unshared them.
  1191. for folderID, folderCfg := range folders {
  1192. for k := 0; k < len(folderCfg.Devices); k++ {
  1193. if folderCfg.Devices[k].IntroducedBy != introducerCfg.DeviceID {
  1194. devicesNotIntroduced[folderCfg.Devices[k].DeviceID] = struct{}{}
  1195. continue
  1196. }
  1197. if !foldersDevices.has(folderCfg.Devices[k].DeviceID, folderCfg.ID) {
  1198. // We could not find that folder shared on the
  1199. // introducer with the device that was introduced to us.
  1200. // We should follow and unshare as well.
  1201. l.Infof("Unsharing folder %s with %v as introducer %v no longer shares the folder with that device", folderCfg.Description(), folderCfg.Devices[k].DeviceID, folderCfg.Devices[k].IntroducedBy)
  1202. folderCfg.Devices = append(folderCfg.Devices[:k], folderCfg.Devices[k+1:]...)
  1203. folders[folderID] = folderCfg
  1204. k--
  1205. changed = true
  1206. }
  1207. }
  1208. }
  1209. // Check if we should remove some devices, if the introducer no longer
  1210. // shares any folder with them. Yet do not remove if we share other
  1211. // folders that haven't been introduced by the introducer.
  1212. for deviceID, device := range devices {
  1213. if device.IntroducedBy == introducerCfg.DeviceID {
  1214. if !foldersDevices.hasDevice(deviceID) {
  1215. if _, ok := devicesNotIntroduced[deviceID]; !ok {
  1216. // The introducer no longer shares any folder with the
  1217. // device, remove the device.
  1218. l.Infof("Removing device %v as introducer %v no longer shares any folders with that device", deviceID, device.IntroducedBy)
  1219. changed = true
  1220. delete(devices, deviceID)
  1221. continue
  1222. }
  1223. l.Infof("Would have removed %v as %v no longer shares any folders, yet there are other folders that are shared with this device that haven't been introduced by this introducer.", deviceID, device.IntroducedBy)
  1224. }
  1225. }
  1226. }
  1227. return folders, devices, changed
  1228. }
  1229. // handleAutoAccepts handles adding and sharing folders for devices that have
  1230. // AutoAcceptFolders set to true.
  1231. func (m *model) handleAutoAccepts(deviceCfg config.DeviceConfiguration, folder protocol.Folder) bool {
  1232. if cfg, ok := m.cfg.Folder(folder.ID); !ok {
  1233. defaultPath := m.cfg.Options().DefaultFolderPath
  1234. defaultPathFs := fs.NewFilesystem(fs.FilesystemTypeBasic, defaultPath)
  1235. pathAlternatives := []string{
  1236. sanitizePath(folder.Label),
  1237. sanitizePath(folder.ID),
  1238. }
  1239. for _, path := range pathAlternatives {
  1240. if _, err := defaultPathFs.Lstat(path); !fs.IsNotExist(err) {
  1241. continue
  1242. }
  1243. fcfg := config.NewFolderConfiguration(m.id, folder.ID, folder.Label, fs.FilesystemTypeBasic, filepath.Join(defaultPath, path))
  1244. fcfg.Devices = append(fcfg.Devices, config.FolderDeviceConfiguration{
  1245. DeviceID: deviceCfg.DeviceID,
  1246. })
  1247. // Need to wait for the waiter, as this calls CommitConfiguration,
  1248. // which sets up the folder and as we return from this call,
  1249. // ClusterConfig starts poking at m.folderFiles and other things
  1250. // that might not exist until the config is committed.
  1251. w, _ := m.cfg.SetFolder(fcfg)
  1252. w.Wait()
  1253. l.Infof("Auto-accepted %s folder %s at path %s", deviceCfg.DeviceID, folder.Description(), fcfg.Path)
  1254. return true
  1255. }
  1256. l.Infof("Failed to auto-accept folder %s from %s due to path conflict", folder.Description(), deviceCfg.DeviceID)
  1257. return false
  1258. } else {
  1259. for _, device := range cfg.DeviceIDs() {
  1260. if device == deviceCfg.DeviceID {
  1261. // Already shared nothing todo.
  1262. return false
  1263. }
  1264. }
  1265. cfg.Devices = append(cfg.Devices, config.FolderDeviceConfiguration{
  1266. DeviceID: deviceCfg.DeviceID,
  1267. })
  1268. w, _ := m.cfg.SetFolder(cfg)
  1269. w.Wait()
  1270. l.Infof("Shared %s with %s due to auto-accept", folder.ID, deviceCfg.DeviceID)
  1271. return true
  1272. }
  1273. }
  1274. func (m *model) introduceDevice(device protocol.Device, introducerCfg config.DeviceConfiguration) config.DeviceConfiguration {
  1275. addresses := []string{"dynamic"}
  1276. for _, addr := range device.Addresses {
  1277. if addr != "dynamic" {
  1278. addresses = append(addresses, addr)
  1279. }
  1280. }
  1281. l.Infof("Adding device %v to config (vouched for by introducer %v)", device.ID, introducerCfg.DeviceID)
  1282. newDeviceCfg := config.DeviceConfiguration{
  1283. DeviceID: device.ID,
  1284. Name: device.Name,
  1285. Compression: introducerCfg.Compression,
  1286. Addresses: addresses,
  1287. CertName: device.CertName,
  1288. IntroducedBy: introducerCfg.DeviceID,
  1289. }
  1290. // The introducers' introducers are also our introducers.
  1291. if device.Introducer {
  1292. l.Infof("Device %v is now also an introducer", device.ID)
  1293. newDeviceCfg.Introducer = true
  1294. newDeviceCfg.SkipIntroductionRemovals = device.SkipIntroductionRemovals
  1295. }
  1296. return newDeviceCfg
  1297. }
  1298. // Closed is called when a connection has been closed
  1299. func (m *model) Closed(conn protocol.Connection, err error) {
  1300. device := conn.ID()
  1301. m.pmut.Lock()
  1302. conn, ok := m.conn[device]
  1303. if !ok {
  1304. m.pmut.Unlock()
  1305. return
  1306. }
  1307. delete(m.conn, device)
  1308. delete(m.connRequestLimiters, device)
  1309. delete(m.helloMessages, device)
  1310. delete(m.deviceDownloads, device)
  1311. delete(m.remotePausedFolders, device)
  1312. closed := m.closed[device]
  1313. delete(m.closed, device)
  1314. m.pmut.Unlock()
  1315. m.progressEmitter.temporaryIndexUnsubscribe(conn)
  1316. l.Infof("Connection to %s at %s closed: %v", device, conn.Name(), err)
  1317. m.evLogger.Log(events.DeviceDisconnected, map[string]string{
  1318. "id": device.String(),
  1319. "error": err.Error(),
  1320. })
  1321. close(closed)
  1322. }
  1323. // closeConns will close the underlying connection for given devices and return
  1324. // a waiter that will return once all the connections are finished closing.
  1325. func (m *model) closeConns(devs []protocol.DeviceID, err error) config.Waiter {
  1326. conns := make([]connections.Connection, 0, len(devs))
  1327. closed := make([]chan struct{}, 0, len(devs))
  1328. m.pmut.RLock()
  1329. for _, dev := range devs {
  1330. if conn, ok := m.conn[dev]; ok {
  1331. conns = append(conns, conn)
  1332. closed = append(closed, m.closed[dev])
  1333. }
  1334. }
  1335. m.pmut.RUnlock()
  1336. for _, conn := range conns {
  1337. conn.Close(err)
  1338. }
  1339. return &channelWaiter{chans: closed}
  1340. }
  1341. // closeConn closes the underlying connection for the given device and returns
  1342. // a waiter that will return once the connection is finished closing.
  1343. func (m *model) closeConn(dev protocol.DeviceID, err error) config.Waiter {
  1344. return m.closeConns([]protocol.DeviceID{dev}, err)
  1345. }
  1346. type channelWaiter struct {
  1347. chans []chan struct{}
  1348. }
  1349. func (w *channelWaiter) Wait() {
  1350. for _, c := range w.chans {
  1351. <-c
  1352. }
  1353. }
  1354. // Implements protocol.RequestResponse
  1355. type requestResponse struct {
  1356. data []byte
  1357. closed chan struct{}
  1358. once stdsync.Once
  1359. }
  1360. func newRequestResponse(size int) *requestResponse {
  1361. return &requestResponse{
  1362. data: protocol.BufferPool.Get(size),
  1363. closed: make(chan struct{}),
  1364. }
  1365. }
  1366. func (r *requestResponse) Data() []byte {
  1367. return r.data
  1368. }
  1369. func (r *requestResponse) Close() {
  1370. r.once.Do(func() {
  1371. protocol.BufferPool.Put(r.data)
  1372. close(r.closed)
  1373. })
  1374. }
  1375. func (r *requestResponse) Wait() {
  1376. <-r.closed
  1377. }
  1378. // Request returns the specified data segment by reading it from local disk.
  1379. // Implements the protocol.Model interface.
  1380. func (m *model) Request(deviceID protocol.DeviceID, folder, name string, size int32, offset int64, hash []byte, weakHash uint32, fromTemporary bool) (out protocol.RequestResponse, err error) {
  1381. if size < 0 || offset < 0 {
  1382. return nil, protocol.ErrInvalid
  1383. }
  1384. m.fmut.RLock()
  1385. folderCfg, ok := m.folderCfgs[folder]
  1386. folderIgnores := m.folderIgnores[folder]
  1387. m.fmut.RUnlock()
  1388. if !ok {
  1389. // The folder might be already unpaused in the config, but not yet
  1390. // in the model.
  1391. l.Debugf("Request from %s for file %s in unstarted folder %q", deviceID, name, folder)
  1392. return nil, protocol.ErrGeneric
  1393. }
  1394. if !folderCfg.SharedWith(deviceID) {
  1395. l.Warnf("Request from %s for file %s in unshared folder %q", deviceID, name, folder)
  1396. return nil, protocol.ErrGeneric
  1397. }
  1398. if folderCfg.Paused {
  1399. l.Debugf("Request from %s for file %s in paused folder %q", deviceID, name, folder)
  1400. return nil, protocol.ErrGeneric
  1401. }
  1402. // Make sure the path is valid and in canonical form
  1403. if name, err = fs.Canonicalize(name); err != nil {
  1404. l.Debugf("Request from %s in folder %q for invalid filename %s", deviceID, folder, name)
  1405. return nil, protocol.ErrGeneric
  1406. }
  1407. if deviceID != protocol.LocalDeviceID {
  1408. l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d t=%v", m, deviceID, folder, name, offset, size, fromTemporary)
  1409. }
  1410. if fs.IsInternal(name) {
  1411. l.Debugf("%v REQ(in) for internal file: %s: %q / %q o=%d s=%d", m, deviceID, folder, name, offset, size)
  1412. return nil, protocol.ErrInvalid
  1413. }
  1414. if folderIgnores.Match(name).IsIgnored() {
  1415. l.Debugf("%v REQ(in) for ignored file: %s: %q / %q o=%d s=%d", m, deviceID, folder, name, offset, size)
  1416. return nil, protocol.ErrInvalid
  1417. }
  1418. folderFs := folderCfg.Filesystem()
  1419. if err := osutil.TraversesSymlink(folderFs, filepath.Dir(name)); err != nil {
  1420. l.Debugf("%v REQ(in) traversal check: %s - %s: %q / %q o=%d s=%d", m, err, deviceID, folder, name, offset, size)
  1421. return nil, protocol.ErrNoSuchFile
  1422. }
  1423. // Restrict parallel requests by connection/device
  1424. m.pmut.RLock()
  1425. limiter := m.connRequestLimiters[deviceID]
  1426. m.pmut.RUnlock()
  1427. if limiter != nil {
  1428. limiter.take(int(size))
  1429. }
  1430. // The requestResponse releases the bytes to the limiter when its Close method is called.
  1431. res := newRequestResponse(int(size))
  1432. defer func() {
  1433. // Close it ourselves if it isn't returned due to an error
  1434. if err != nil {
  1435. res.Close()
  1436. }
  1437. }()
  1438. if limiter != nil {
  1439. go func() {
  1440. res.Wait()
  1441. limiter.give(int(size))
  1442. }()
  1443. }
  1444. // Only check temp files if the flag is set, and if we are set to advertise
  1445. // the temp indexes.
  1446. if fromTemporary && !folderCfg.DisableTempIndexes {
  1447. tempFn := fs.TempName(name)
  1448. if info, err := folderFs.Lstat(tempFn); err != nil || !info.IsRegular() {
  1449. // Reject reads for anything that doesn't exist or is something
  1450. // other than a regular file.
  1451. l.Debugf("%v REQ(in) failed stating temp file (%v): %s: %q / %q o=%d s=%d", m, err, deviceID, folder, name, offset, size)
  1452. return nil, protocol.ErrNoSuchFile
  1453. }
  1454. err := readOffsetIntoBuf(folderFs, tempFn, offset, res.data)
  1455. if err == nil && scanner.Validate(res.data, hash, weakHash) {
  1456. return res, nil
  1457. }
  1458. // Fall through to reading from a non-temp file, just incase the temp
  1459. // file has finished downloading.
  1460. }
  1461. if info, err := folderFs.Lstat(name); err != nil || !info.IsRegular() {
  1462. // Reject reads for anything that doesn't exist or is something
  1463. // other than a regular file.
  1464. l.Debugf("%v REQ(in) failed stating file (%v): %s: %q / %q o=%d s=%d", m, err, deviceID, folder, name, offset, size)
  1465. return nil, protocol.ErrNoSuchFile
  1466. }
  1467. if err := readOffsetIntoBuf(folderFs, name, offset, res.data); fs.IsNotExist(err) {
  1468. l.Debugf("%v REQ(in) file doesn't exist: %s: %q / %q o=%d s=%d", m, deviceID, folder, name, offset, size)
  1469. return nil, protocol.ErrNoSuchFile
  1470. } else if err != nil {
  1471. l.Debugf("%v REQ(in) failed reading file (%v): %s: %q / %q o=%d s=%d", m, err, deviceID, folder, name, offset, size)
  1472. return nil, protocol.ErrGeneric
  1473. }
  1474. if !scanner.Validate(res.data, hash, weakHash) {
  1475. m.recheckFile(deviceID, folderFs, folder, name, size, offset, hash)
  1476. l.Debugf("%v REQ(in) failed validating data (%v): %s: %q / %q o=%d s=%d", m, err, deviceID, folder, name, offset, size)
  1477. return nil, protocol.ErrNoSuchFile
  1478. }
  1479. return res, nil
  1480. }
  1481. func (m *model) recheckFile(deviceID protocol.DeviceID, folderFs fs.Filesystem, folder, name string, size int32, offset int64, hash []byte) {
  1482. cf, ok := m.CurrentFolderFile(folder, name)
  1483. if !ok {
  1484. l.Debugf("%v recheckFile: %s: %q / %q: no current file", m, deviceID, folder, name)
  1485. return
  1486. }
  1487. if cf.IsDeleted() || cf.IsInvalid() || cf.IsSymlink() || cf.IsDirectory() {
  1488. l.Debugf("%v recheckFile: %s: %q / %q: not a regular file", m, deviceID, folder, name)
  1489. return
  1490. }
  1491. blockIndex := int(offset / int64(cf.BlockSize()))
  1492. if blockIndex >= len(cf.Blocks) {
  1493. l.Debugf("%v recheckFile: %s: %q / %q i=%d: block index too far", m, deviceID, folder, name, blockIndex)
  1494. return
  1495. }
  1496. block := cf.Blocks[blockIndex]
  1497. // Seems to want a different version of the file, whatever.
  1498. if !bytes.Equal(block.Hash, hash) {
  1499. l.Debugf("%v recheckFile: %s: %q / %q i=%d: hash mismatch %x != %x", m, deviceID, folder, name, blockIndex, block.Hash, hash)
  1500. return
  1501. }
  1502. // The hashes provided part of the request match what we expect to find according
  1503. // to what we have in the database, yet the content we've read off the filesystem doesn't
  1504. // Something is fishy, invalidate the file and rescan it.
  1505. // The file will temporarily become invalid, which is ok as the content is messed up.
  1506. m.fmut.RLock()
  1507. runner, ok := m.folderRunners[folder]
  1508. m.fmut.RUnlock()
  1509. if !ok {
  1510. l.Debugf("%v recheckFile: %s: %q / %q: Folder stopped before rescan could be scheduled", m, deviceID, folder, name)
  1511. return
  1512. }
  1513. if err := runner.ForceRescan(cf); err != nil {
  1514. l.Debugf("%v recheckFile: %s: %q / %q rescan: %s", m, deviceID, folder, name, err)
  1515. return
  1516. }
  1517. l.Debugf("%v recheckFile: %s: %q / %q", m, deviceID, folder, name)
  1518. }
  1519. func (m *model) CurrentFolderFile(folder string, file string) (protocol.FileInfo, bool) {
  1520. m.fmut.RLock()
  1521. fs, ok := m.folderFiles[folder]
  1522. m.fmut.RUnlock()
  1523. if !ok {
  1524. return protocol.FileInfo{}, false
  1525. }
  1526. return fs.Get(protocol.LocalDeviceID, file)
  1527. }
  1528. func (m *model) CurrentGlobalFile(folder string, file string) (protocol.FileInfo, bool) {
  1529. m.fmut.RLock()
  1530. fs, ok := m.folderFiles[folder]
  1531. m.fmut.RUnlock()
  1532. if !ok {
  1533. return protocol.FileInfo{}, false
  1534. }
  1535. return fs.GetGlobal(file)
  1536. }
  1537. // Connection returns the current connection for device, and a boolean whether a connection was found.
  1538. func (m *model) Connection(deviceID protocol.DeviceID) (connections.Connection, bool) {
  1539. m.pmut.RLock()
  1540. cn, ok := m.conn[deviceID]
  1541. m.pmut.RUnlock()
  1542. if ok {
  1543. m.deviceWasSeen(deviceID)
  1544. }
  1545. return cn, ok
  1546. }
  1547. func (m *model) GetIgnores(folder string) ([]string, []string, error) {
  1548. m.fmut.RLock()
  1549. cfg, cfgOk := m.folderCfgs[folder]
  1550. ignores, ignoresOk := m.folderIgnores[folder]
  1551. m.fmut.RUnlock()
  1552. if !cfgOk {
  1553. cfg, cfgOk = m.cfg.Folders()[folder]
  1554. if !cfgOk {
  1555. return nil, nil, fmt.Errorf("Folder %s does not exist", folder)
  1556. }
  1557. }
  1558. // On creation a new folder with ignore patterns validly has no marker yet.
  1559. if err := cfg.CheckPath(); err != nil && err != config.ErrMarkerMissing {
  1560. return nil, nil, err
  1561. }
  1562. if !ignoresOk {
  1563. ignores = ignore.New(fs.NewFilesystem(cfg.FilesystemType, cfg.Path))
  1564. }
  1565. if err := ignores.Load(".stignore"); err != nil && !fs.IsNotExist(err) {
  1566. return nil, nil, err
  1567. }
  1568. return ignores.Lines(), ignores.Patterns(), nil
  1569. }
  1570. func (m *model) SetIgnores(folder string, content []string) error {
  1571. cfg, ok := m.cfg.Folders()[folder]
  1572. if !ok {
  1573. return fmt.Errorf("folder %s does not exist", cfg.Description())
  1574. }
  1575. err := cfg.CheckPath()
  1576. if err == config.ErrPathMissing {
  1577. if err = cfg.CreateRoot(); err != nil {
  1578. return errors.Wrap(err, "failed to create folder root")
  1579. }
  1580. err = cfg.CheckPath()
  1581. }
  1582. if err != nil && err != config.ErrMarkerMissing {
  1583. return err
  1584. }
  1585. if err := ignore.WriteIgnores(cfg.Filesystem(), ".stignore", content); err != nil {
  1586. l.Warnln("Saving .stignore:", err)
  1587. return err
  1588. }
  1589. m.fmut.RLock()
  1590. runner, ok := m.folderRunners[folder]
  1591. m.fmut.RUnlock()
  1592. if ok {
  1593. return runner.Scan(nil)
  1594. }
  1595. return nil
  1596. }
  1597. // OnHello is called when an device connects to us.
  1598. // This allows us to extract some information from the Hello message
  1599. // and add it to a list of known devices ahead of any checks.
  1600. func (m *model) OnHello(remoteID protocol.DeviceID, addr net.Addr, hello protocol.HelloResult) error {
  1601. if m.cfg.IgnoredDevice(remoteID) {
  1602. return errDeviceIgnored
  1603. }
  1604. cfg, ok := m.cfg.Device(remoteID)
  1605. if !ok {
  1606. m.cfg.AddOrUpdatePendingDevice(remoteID, hello.DeviceName, addr.String())
  1607. _ = m.cfg.Save() // best effort
  1608. m.evLogger.Log(events.DeviceRejected, map[string]string{
  1609. "name": hello.DeviceName,
  1610. "device": remoteID.String(),
  1611. "address": addr.String(),
  1612. })
  1613. return errDeviceUnknown
  1614. }
  1615. if cfg.Paused {
  1616. return errDevicePaused
  1617. }
  1618. if len(cfg.AllowedNetworks) > 0 {
  1619. if !connections.IsAllowedNetwork(addr.String(), cfg.AllowedNetworks) {
  1620. return errNetworkNotAllowed
  1621. }
  1622. }
  1623. return nil
  1624. }
  1625. // GetHello is called when we are about to connect to some remote device.
  1626. func (m *model) GetHello(id protocol.DeviceID) protocol.HelloIntf {
  1627. name := ""
  1628. if _, ok := m.cfg.Device(id); ok {
  1629. name = m.cfg.MyName()
  1630. }
  1631. return &protocol.Hello{
  1632. DeviceName: name,
  1633. ClientName: m.clientName,
  1634. ClientVersion: m.clientVersion,
  1635. }
  1636. }
  1637. // AddConnection adds a new peer connection to the model. An initial index will
  1638. // be sent to the connected peer, thereafter index updates whenever the local
  1639. // folder changes.
  1640. func (m *model) AddConnection(conn connections.Connection, hello protocol.HelloResult) {
  1641. deviceID := conn.ID()
  1642. device, ok := m.cfg.Device(deviceID)
  1643. if !ok {
  1644. l.Infoln("Trying to add connection to unknown device")
  1645. return
  1646. }
  1647. m.pmut.Lock()
  1648. if oldConn, ok := m.conn[deviceID]; ok {
  1649. l.Infoln("Replacing old connection", oldConn, "with", conn, "for", deviceID)
  1650. // There is an existing connection to this device that we are
  1651. // replacing. We must close the existing connection and wait for the
  1652. // close to complete before adding the new connection. We do the
  1653. // actual close without holding pmut as the connection will call
  1654. // back into Closed() for the cleanup.
  1655. closed := m.closed[deviceID]
  1656. m.pmut.Unlock()
  1657. oldConn.Close(errReplacingConnection)
  1658. <-closed
  1659. m.pmut.Lock()
  1660. }
  1661. m.conn[deviceID] = conn
  1662. m.closed[deviceID] = make(chan struct{})
  1663. m.deviceDownloads[deviceID] = newDeviceDownloadState()
  1664. // 0: default, <0: no limiting
  1665. switch {
  1666. case device.MaxRequestKiB > 0:
  1667. m.connRequestLimiters[deviceID] = newByteSemaphore(1024 * device.MaxRequestKiB)
  1668. case device.MaxRequestKiB == 0:
  1669. m.connRequestLimiters[deviceID] = newByteSemaphore(1024 * defaultPullerPendingKiB)
  1670. }
  1671. m.helloMessages[deviceID] = hello
  1672. event := map[string]string{
  1673. "id": deviceID.String(),
  1674. "deviceName": hello.DeviceName,
  1675. "clientName": hello.ClientName,
  1676. "clientVersion": hello.ClientVersion,
  1677. "type": conn.Type(),
  1678. }
  1679. addr := conn.RemoteAddr()
  1680. if addr != nil {
  1681. event["addr"] = addr.String()
  1682. }
  1683. m.evLogger.Log(events.DeviceConnected, event)
  1684. l.Infof(`Device %s client is "%s %s" named "%s" at %s`, deviceID, hello.ClientName, hello.ClientVersion, hello.DeviceName, conn)
  1685. conn.Start()
  1686. m.pmut.Unlock()
  1687. // Acquires fmut, so has to be done outside of pmut.
  1688. cm := m.generateClusterConfig(deviceID)
  1689. conn.ClusterConfig(cm)
  1690. if (device.Name == "" || m.cfg.Options().OverwriteRemoteDevNames) && hello.DeviceName != "" {
  1691. device.Name = hello.DeviceName
  1692. m.cfg.SetDevice(device)
  1693. m.cfg.Save()
  1694. }
  1695. m.deviceWasSeen(deviceID)
  1696. }
  1697. func (m *model) DownloadProgress(device protocol.DeviceID, folder string, updates []protocol.FileDownloadProgressUpdate) error {
  1698. m.fmut.RLock()
  1699. cfg, ok := m.folderCfgs[folder]
  1700. m.fmut.RUnlock()
  1701. if !ok || cfg.DisableTempIndexes || !cfg.SharedWith(device) {
  1702. return nil
  1703. }
  1704. m.pmut.RLock()
  1705. downloads := m.deviceDownloads[device]
  1706. m.pmut.RUnlock()
  1707. downloads.Update(folder, updates)
  1708. state := downloads.GetBlockCounts(folder)
  1709. m.evLogger.Log(events.RemoteDownloadProgress, map[string]interface{}{
  1710. "device": device.String(),
  1711. "folder": folder,
  1712. "state": state,
  1713. })
  1714. return nil
  1715. }
  1716. func (m *model) deviceWasSeen(deviceID protocol.DeviceID) {
  1717. m.fmut.RLock()
  1718. sr, ok := m.deviceStatRefs[deviceID]
  1719. m.fmut.RUnlock()
  1720. if ok {
  1721. sr.WasSeen()
  1722. }
  1723. }
  1724. type indexSender struct {
  1725. suture.Service
  1726. conn protocol.Connection
  1727. folder string
  1728. dev string
  1729. fset *db.FileSet
  1730. prevSequence int64
  1731. dropSymlinks bool
  1732. evLogger events.Logger
  1733. connClosed chan struct{}
  1734. }
  1735. func (s *indexSender) serve(ctx context.Context) {
  1736. var err error
  1737. l.Debugf("Starting indexSender for %s to %s at %s (slv=%d)", s.folder, s.dev, s.conn, s.prevSequence)
  1738. defer l.Debugf("Exiting indexSender for %s to %s at %s: %v", s.folder, s.dev, s.conn, err)
  1739. // We need to send one index, regardless of whether there is something to send or not
  1740. err = s.sendIndexTo(ctx)
  1741. // Subscribe to LocalIndexUpdated (we have new information to send) and
  1742. // DeviceDisconnected (it might be us who disconnected, so we should
  1743. // exit).
  1744. sub := s.evLogger.Subscribe(events.LocalIndexUpdated | events.DeviceDisconnected)
  1745. defer sub.Unsubscribe()
  1746. evChan := sub.C()
  1747. ticker := time.NewTicker(time.Minute)
  1748. defer ticker.Stop()
  1749. for err == nil {
  1750. select {
  1751. case <-ctx.Done():
  1752. return
  1753. case <-s.connClosed:
  1754. return
  1755. default:
  1756. }
  1757. // While we have sent a sequence at least equal to the one
  1758. // currently in the database, wait for the local index to update. The
  1759. // local index may update for other folders than the one we are
  1760. // sending for.
  1761. if s.fset.Sequence(protocol.LocalDeviceID) <= s.prevSequence {
  1762. select {
  1763. case <-ctx.Done():
  1764. return
  1765. case <-s.connClosed:
  1766. return
  1767. case <-evChan:
  1768. case <-ticker.C:
  1769. }
  1770. continue
  1771. }
  1772. err = s.sendIndexTo(ctx)
  1773. // Wait a short amount of time before entering the next loop. If there
  1774. // are continuous changes happening to the local index, this gives us
  1775. // time to batch them up a little.
  1776. time.Sleep(250 * time.Millisecond)
  1777. }
  1778. }
  1779. // Complete implements the suture.IsCompletable interface. When Serve terminates
  1780. // before Stop is called, the supervisor will check for this method and if it
  1781. // returns true removes the service instead of restarting it. Here it always
  1782. // returns true, as indexSender only terminates when a connection is
  1783. // closed/has failed, in which case retrying doesn't help.
  1784. func (s *indexSender) Complete() bool { return true }
  1785. // sendIndexTo sends file infos with a sequence number higher than prevSequence and
  1786. // returns the highest sent sequence number.
  1787. func (s *indexSender) sendIndexTo(ctx context.Context) error {
  1788. initial := s.prevSequence == 0
  1789. batch := newFileInfoBatch(nil)
  1790. batch.flushFn = func(fs []protocol.FileInfo) error {
  1791. l.Debugf("%v: Sending %d files (<%d bytes)", s, len(batch.infos), batch.size)
  1792. if initial {
  1793. initial = false
  1794. return s.conn.Index(ctx, s.folder, fs)
  1795. }
  1796. return s.conn.IndexUpdate(ctx, s.folder, fs)
  1797. }
  1798. var err error
  1799. var f protocol.FileInfo
  1800. s.fset.WithHaveSequence(s.prevSequence+1, func(fi db.FileIntf) bool {
  1801. if err = batch.flushIfFull(); err != nil {
  1802. return false
  1803. }
  1804. if shouldDebug() {
  1805. if fi.SequenceNo() < s.prevSequence+1 {
  1806. panic(fmt.Sprintln("sequence lower than requested, got:", fi.SequenceNo(), ", asked to start at:", s.prevSequence+1))
  1807. }
  1808. if f.Sequence > 0 && fi.SequenceNo() <= f.Sequence {
  1809. panic(fmt.Sprintln("non-increasing sequence, current:", fi.SequenceNo(), "<= previous:", f.Sequence))
  1810. }
  1811. }
  1812. f = fi.(protocol.FileInfo)
  1813. // Mark the file as invalid if any of the local bad stuff flags are set.
  1814. f.RawInvalid = f.IsInvalid()
  1815. // If the file is marked LocalReceive (i.e., changed locally on a
  1816. // receive only folder) we do not want it to ever become the
  1817. // globally best version, invalid or not.
  1818. if f.IsReceiveOnlyChanged() {
  1819. f.Version = protocol.Vector{}
  1820. }
  1821. f.LocalFlags = 0 // never sent externally
  1822. if s.dropSymlinks && f.IsSymlink() {
  1823. // Do not send index entries with symlinks to clients that can't
  1824. // handle it. Fixes issue #3802. Once both sides are upgraded, a
  1825. // rescan (i.e., change) of the symlink is required for it to
  1826. // sync again, due to delta indexes.
  1827. return true
  1828. }
  1829. batch.append(f)
  1830. return true
  1831. })
  1832. if err != nil {
  1833. return err
  1834. }
  1835. err = batch.flush()
  1836. // True if there was nothing to be sent
  1837. if f.Sequence == 0 {
  1838. return err
  1839. }
  1840. s.prevSequence = f.Sequence
  1841. return err
  1842. }
  1843. func (s *indexSender) String() string {
  1844. return fmt.Sprintf("indexSender@%p for %s to %s at %s", s, s.folder, s.dev, s.conn)
  1845. }
  1846. func (m *model) requestGlobal(ctx context.Context, deviceID protocol.DeviceID, folder, name string, offset int64, size int, hash []byte, weakHash uint32, fromTemporary bool) ([]byte, error) {
  1847. m.pmut.RLock()
  1848. nc, ok := m.conn[deviceID]
  1849. m.pmut.RUnlock()
  1850. if !ok {
  1851. return nil, fmt.Errorf("requestGlobal: no such device: %s", deviceID)
  1852. }
  1853. l.Debugf("%v REQ(out): %s: %q / %q o=%d s=%d h=%x wh=%x ft=%t", m, deviceID, folder, name, offset, size, hash, weakHash, fromTemporary)
  1854. return nc.Request(ctx, folder, name, offset, size, hash, weakHash, fromTemporary)
  1855. }
  1856. func (m *model) ScanFolders() map[string]error {
  1857. m.fmut.RLock()
  1858. folders := make([]string, 0, len(m.folderCfgs))
  1859. for folder := range m.folderCfgs {
  1860. folders = append(folders, folder)
  1861. }
  1862. m.fmut.RUnlock()
  1863. errors := make(map[string]error, len(m.folderCfgs))
  1864. errorsMut := sync.NewMutex()
  1865. wg := sync.NewWaitGroup()
  1866. wg.Add(len(folders))
  1867. for _, folder := range folders {
  1868. folder := folder
  1869. go func() {
  1870. err := m.ScanFolder(folder)
  1871. if err != nil {
  1872. errorsMut.Lock()
  1873. errors[folder] = err
  1874. errorsMut.Unlock()
  1875. }
  1876. wg.Done()
  1877. }()
  1878. }
  1879. wg.Wait()
  1880. return errors
  1881. }
  1882. func (m *model) ScanFolder(folder string) error {
  1883. return m.ScanFolderSubdirs(folder, nil)
  1884. }
  1885. func (m *model) ScanFolderSubdirs(folder string, subs []string) error {
  1886. m.fmut.RLock()
  1887. err := m.checkFolderRunningLocked(folder)
  1888. runner := m.folderRunners[folder]
  1889. m.fmut.RUnlock()
  1890. if err != nil {
  1891. return err
  1892. }
  1893. return runner.Scan(subs)
  1894. }
  1895. func (m *model) DelayScan(folder string, next time.Duration) {
  1896. m.fmut.RLock()
  1897. runner, ok := m.folderRunners[folder]
  1898. m.fmut.RUnlock()
  1899. if !ok {
  1900. return
  1901. }
  1902. runner.DelayScan(next)
  1903. }
  1904. // numHashers returns the number of hasher routines to use for a given folder,
  1905. // taking into account configuration and available CPU cores.
  1906. func (m *model) numHashers(folder string) int {
  1907. m.fmut.RLock()
  1908. folderCfg := m.folderCfgs[folder]
  1909. numFolders := len(m.folderCfgs)
  1910. m.fmut.RUnlock()
  1911. if folderCfg.Hashers > 0 {
  1912. // Specific value set in the config, use that.
  1913. return folderCfg.Hashers
  1914. }
  1915. if runtime.GOOS == "windows" || runtime.GOOS == "darwin" {
  1916. // Interactive operating systems; don't load the system too heavily by
  1917. // default.
  1918. return 1
  1919. }
  1920. // For other operating systems and architectures, lets try to get some
  1921. // work done... Divide the available CPU cores among the configured
  1922. // folders.
  1923. if perFolder := runtime.GOMAXPROCS(-1) / numFolders; perFolder > 0 {
  1924. return perFolder
  1925. }
  1926. return 1
  1927. }
  1928. // generateClusterConfig returns a ClusterConfigMessage that is correct for
  1929. // the given peer device
  1930. func (m *model) generateClusterConfig(device protocol.DeviceID) protocol.ClusterConfig {
  1931. var message protocol.ClusterConfig
  1932. m.fmut.RLock()
  1933. defer m.fmut.RUnlock()
  1934. for _, folderCfg := range m.cfg.FolderList() {
  1935. if !folderCfg.SharedWith(device) {
  1936. continue
  1937. }
  1938. protocolFolder := protocol.Folder{
  1939. ID: folderCfg.ID,
  1940. Label: folderCfg.Label,
  1941. ReadOnly: folderCfg.Type == config.FolderTypeSendOnly,
  1942. IgnorePermissions: folderCfg.IgnorePerms,
  1943. IgnoreDelete: folderCfg.IgnoreDelete,
  1944. DisableTempIndexes: folderCfg.DisableTempIndexes,
  1945. Paused: folderCfg.Paused,
  1946. }
  1947. var fs *db.FileSet
  1948. if !folderCfg.Paused {
  1949. fs = m.folderFiles[folderCfg.ID]
  1950. }
  1951. for _, device := range folderCfg.Devices {
  1952. deviceCfg, _ := m.cfg.Device(device.DeviceID)
  1953. protocolDevice := protocol.Device{
  1954. ID: deviceCfg.DeviceID,
  1955. Name: deviceCfg.Name,
  1956. Addresses: deviceCfg.Addresses,
  1957. Compression: deviceCfg.Compression,
  1958. CertName: deviceCfg.CertName,
  1959. Introducer: deviceCfg.Introducer,
  1960. }
  1961. if fs != nil {
  1962. if deviceCfg.DeviceID == m.id {
  1963. protocolDevice.IndexID = fs.IndexID(protocol.LocalDeviceID)
  1964. protocolDevice.MaxSequence = fs.Sequence(protocol.LocalDeviceID)
  1965. } else {
  1966. protocolDevice.IndexID = fs.IndexID(deviceCfg.DeviceID)
  1967. protocolDevice.MaxSequence = fs.Sequence(deviceCfg.DeviceID)
  1968. }
  1969. }
  1970. protocolFolder.Devices = append(protocolFolder.Devices, protocolDevice)
  1971. }
  1972. message.Folders = append(message.Folders, protocolFolder)
  1973. }
  1974. return message
  1975. }
  1976. func (m *model) State(folder string) (string, time.Time, error) {
  1977. m.fmut.RLock()
  1978. runner, ok := m.folderRunners[folder]
  1979. m.fmut.RUnlock()
  1980. if !ok {
  1981. // The returned error should be an actual folder error, so returning
  1982. // errors.New("does not exist") or similar here would be
  1983. // inappropriate.
  1984. return "", time.Time{}, nil
  1985. }
  1986. state, changed, err := runner.getState()
  1987. return state.String(), changed, err
  1988. }
  1989. func (m *model) FolderErrors(folder string) ([]FileError, error) {
  1990. m.fmut.RLock()
  1991. err := m.checkFolderRunningLocked(folder)
  1992. runner := m.folderRunners[folder]
  1993. m.fmut.RUnlock()
  1994. if err != nil {
  1995. return nil, err
  1996. }
  1997. return runner.Errors(), nil
  1998. }
  1999. func (m *model) WatchError(folder string) error {
  2000. m.fmut.RLock()
  2001. err := m.checkFolderRunningLocked(folder)
  2002. runner := m.folderRunners[folder]
  2003. m.fmut.RUnlock()
  2004. if err != nil {
  2005. return nil // If the folder isn't running, there's no error to report.
  2006. }
  2007. return runner.WatchError()
  2008. }
  2009. func (m *model) Override(folder string) {
  2010. // Grab the runner and the file set.
  2011. m.fmut.RLock()
  2012. runner, ok := m.folderRunners[folder]
  2013. m.fmut.RUnlock()
  2014. if !ok {
  2015. return
  2016. }
  2017. // Run the override, taking updates as if they came from scanning.
  2018. runner.Override()
  2019. }
  2020. func (m *model) Revert(folder string) {
  2021. // Grab the runner and the file set.
  2022. m.fmut.RLock()
  2023. runner, ok := m.folderRunners[folder]
  2024. m.fmut.RUnlock()
  2025. if !ok {
  2026. return
  2027. }
  2028. // Run the revert, taking updates as if they came from scanning.
  2029. runner.Revert()
  2030. }
  2031. // CurrentSequence returns the change version for the given folder.
  2032. // This is guaranteed to increment if the contents of the local folder has
  2033. // changed.
  2034. func (m *model) CurrentSequence(folder string) (int64, bool) {
  2035. m.fmut.RLock()
  2036. fs, ok := m.folderFiles[folder]
  2037. m.fmut.RUnlock()
  2038. if !ok {
  2039. // The folder might not exist, since this can be called with a user
  2040. // specified folder name from the REST interface.
  2041. return 0, false
  2042. }
  2043. return fs.Sequence(protocol.LocalDeviceID), true
  2044. }
  2045. // RemoteSequence returns the change version for the given folder, as
  2046. // sent by remote peers. This is guaranteed to increment if the contents of
  2047. // the remote or global folder has changed.
  2048. func (m *model) RemoteSequence(folder string) (int64, bool) {
  2049. m.fmut.RLock()
  2050. fs, ok := m.folderFiles[folder]
  2051. cfg := m.folderCfgs[folder]
  2052. m.fmut.RUnlock()
  2053. if !ok {
  2054. // The folder might not exist, since this can be called with a user
  2055. // specified folder name from the REST interface.
  2056. return 0, false
  2057. }
  2058. var ver int64
  2059. for _, device := range cfg.Devices {
  2060. ver += fs.Sequence(device.DeviceID)
  2061. }
  2062. return ver, true
  2063. }
  2064. func (m *model) GlobalDirectoryTree(folder, prefix string, levels int, dirsonly bool) map[string]interface{} {
  2065. m.fmut.RLock()
  2066. files, ok := m.folderFiles[folder]
  2067. m.fmut.RUnlock()
  2068. if !ok {
  2069. return nil
  2070. }
  2071. output := make(map[string]interface{})
  2072. sep := string(filepath.Separator)
  2073. prefix = osutil.NativeFilename(prefix)
  2074. if prefix != "" && !strings.HasSuffix(prefix, sep) {
  2075. prefix = prefix + sep
  2076. }
  2077. files.WithPrefixedGlobalTruncated(prefix, func(fi db.FileIntf) bool {
  2078. f := fi.(db.FileInfoTruncated)
  2079. // Don't include the prefix itself.
  2080. if f.IsInvalid() || f.IsDeleted() || strings.HasPrefix(prefix, f.Name) {
  2081. return true
  2082. }
  2083. f.Name = strings.Replace(f.Name, prefix, "", 1)
  2084. var dir, base string
  2085. if f.IsDirectory() && !f.IsSymlink() {
  2086. dir = f.Name
  2087. } else {
  2088. dir = filepath.Dir(f.Name)
  2089. base = filepath.Base(f.Name)
  2090. }
  2091. if levels > -1 && strings.Count(f.Name, sep) > levels {
  2092. return true
  2093. }
  2094. last := output
  2095. if dir != "." {
  2096. for _, path := range strings.Split(dir, sep) {
  2097. directory, ok := last[path]
  2098. if !ok {
  2099. newdir := make(map[string]interface{})
  2100. last[path] = newdir
  2101. last = newdir
  2102. } else {
  2103. last = directory.(map[string]interface{})
  2104. }
  2105. }
  2106. }
  2107. if !dirsonly && base != "" {
  2108. last[base] = []interface{}{
  2109. f.ModTime(), f.FileSize(),
  2110. }
  2111. }
  2112. return true
  2113. })
  2114. return output
  2115. }
  2116. func (m *model) GetFolderVersions(folder string) (map[string][]versioner.FileVersion, error) {
  2117. m.fmut.RLock()
  2118. ver, ok := m.folderVersioners[folder]
  2119. m.fmut.RUnlock()
  2120. if !ok {
  2121. return nil, errFolderMissing
  2122. }
  2123. if ver == nil {
  2124. return nil, errNoVersioner
  2125. }
  2126. return ver.GetVersions()
  2127. }
  2128. func (m *model) RestoreFolderVersions(folder string, versions map[string]time.Time) (map[string]string, error) {
  2129. fcfg, ok := m.cfg.Folder(folder)
  2130. if !ok {
  2131. return nil, errFolderMissing
  2132. }
  2133. m.fmut.RLock()
  2134. ver := m.folderVersioners[folder]
  2135. m.fmut.RUnlock()
  2136. if !ok {
  2137. return nil, errFolderMissing
  2138. }
  2139. if ver == nil {
  2140. return nil, errNoVersioner
  2141. }
  2142. restoreErrors := make(map[string]string)
  2143. for file, version := range versions {
  2144. if err := ver.Restore(file, version); err != nil {
  2145. restoreErrors[file] = err.Error()
  2146. }
  2147. }
  2148. // Trigger scan
  2149. if !fcfg.FSWatcherEnabled {
  2150. go func() { _ = m.ScanFolder(folder) }()
  2151. }
  2152. return restoreErrors, nil
  2153. }
  2154. func (m *model) Availability(folder string, file protocol.FileInfo, block protocol.BlockInfo) []Availability {
  2155. // The slightly unusual locking sequence here is because we need to hold
  2156. // pmut for the duration (as the value returned from foldersFiles can
  2157. // get heavily modified on Close()), but also must acquire fmut before
  2158. // pmut. (The locks can be *released* in any order.)
  2159. m.fmut.RLock()
  2160. m.pmut.RLock()
  2161. defer m.pmut.RUnlock()
  2162. fs, ok := m.folderFiles[folder]
  2163. cfg := m.folderCfgs[folder]
  2164. m.fmut.RUnlock()
  2165. if !ok {
  2166. return nil
  2167. }
  2168. var availabilities []Availability
  2169. next:
  2170. for _, device := range fs.Availability(file.Name) {
  2171. for _, pausedFolder := range m.remotePausedFolders[device] {
  2172. if pausedFolder == folder {
  2173. continue next
  2174. }
  2175. }
  2176. _, ok := m.conn[device]
  2177. if ok {
  2178. availabilities = append(availabilities, Availability{ID: device, FromTemporary: false})
  2179. }
  2180. }
  2181. for _, device := range cfg.Devices {
  2182. if m.deviceDownloads[device.DeviceID].Has(folder, file.Name, file.Version, int32(block.Offset/int64(file.BlockSize()))) {
  2183. availabilities = append(availabilities, Availability{ID: device.DeviceID, FromTemporary: true})
  2184. }
  2185. }
  2186. return availabilities
  2187. }
  2188. // BringToFront bumps the given files priority in the job queue.
  2189. func (m *model) BringToFront(folder, file string) {
  2190. m.fmut.RLock()
  2191. runner, ok := m.folderRunners[folder]
  2192. m.fmut.RUnlock()
  2193. if ok {
  2194. runner.BringToFront(file)
  2195. }
  2196. }
  2197. func (m *model) ResetFolder(folder string) {
  2198. l.Infof("Cleaning data for folder %q", folder)
  2199. db.DropFolder(m.db, folder)
  2200. }
  2201. func (m *model) String() string {
  2202. return fmt.Sprintf("model@%p", m)
  2203. }
  2204. func (m *model) VerifyConfiguration(from, to config.Configuration) error {
  2205. return nil
  2206. }
  2207. func (m *model) CommitConfiguration(from, to config.Configuration) bool {
  2208. // TODO: This should not use reflect, and should take more care to try to handle stuff without restart.
  2209. // Go through the folder configs and figure out if we need to restart or not.
  2210. fromFolders := mapFolders(from.Folders)
  2211. toFolders := mapFolders(to.Folders)
  2212. for folderID, cfg := range toFolders {
  2213. if _, ok := fromFolders[folderID]; !ok {
  2214. // A folder was added.
  2215. if cfg.Paused {
  2216. l.Infoln("Paused folder", cfg.Description())
  2217. } else {
  2218. l.Infoln("Adding folder", cfg.Description())
  2219. m.newFolder(cfg)
  2220. }
  2221. }
  2222. }
  2223. for folderID, fromCfg := range fromFolders {
  2224. toCfg, ok := toFolders[folderID]
  2225. if !ok {
  2226. // The folder was removed.
  2227. m.removeFolder(fromCfg)
  2228. continue
  2229. }
  2230. if fromCfg.Paused && toCfg.Paused {
  2231. continue
  2232. }
  2233. // This folder exists on both sides. Settings might have changed.
  2234. // Check if anything differs that requires a restart.
  2235. if !reflect.DeepEqual(fromCfg.RequiresRestartOnly(), toCfg.RequiresRestartOnly()) {
  2236. m.restartFolder(fromCfg, toCfg)
  2237. }
  2238. // Emit the folder pause/resume event
  2239. if fromCfg.Paused != toCfg.Paused {
  2240. eventType := events.FolderResumed
  2241. if toCfg.Paused {
  2242. eventType = events.FolderPaused
  2243. }
  2244. m.evLogger.Log(eventType, map[string]string{"id": toCfg.ID, "label": toCfg.Label})
  2245. }
  2246. }
  2247. // Removing a device. We actually don't need to do anything.
  2248. // Because folder config has changed (since the device lists do not match)
  2249. // Folders for that had device got "restarted", which involves killing
  2250. // connections to all devices that we were sharing the folder with.
  2251. // At some point model.Close() will get called for that device which will
  2252. // clean residue device state that is not part of any folder.
  2253. // Pausing a device, unpausing is handled by the connection service.
  2254. fromDevices := from.DeviceMap()
  2255. toDevices := to.DeviceMap()
  2256. for deviceID, toCfg := range toDevices {
  2257. fromCfg, ok := fromDevices[deviceID]
  2258. if !ok {
  2259. sr := stats.NewDeviceStatisticsReference(m.db, deviceID.String())
  2260. m.fmut.Lock()
  2261. m.deviceStatRefs[deviceID] = sr
  2262. m.fmut.Unlock()
  2263. continue
  2264. }
  2265. delete(fromDevices, deviceID)
  2266. if fromCfg.Paused == toCfg.Paused {
  2267. continue
  2268. }
  2269. // Ignored folder was removed, reconnect to retrigger the prompt.
  2270. if len(fromCfg.IgnoredFolders) > len(toCfg.IgnoredFolders) {
  2271. m.closeConn(deviceID, errIgnoredFolderRemoved)
  2272. }
  2273. if toCfg.Paused {
  2274. l.Infoln("Pausing", deviceID)
  2275. m.closeConn(deviceID, errDevicePaused)
  2276. m.evLogger.Log(events.DevicePaused, map[string]string{"device": deviceID.String()})
  2277. } else {
  2278. m.evLogger.Log(events.DeviceResumed, map[string]string{"device": deviceID.String()})
  2279. }
  2280. }
  2281. m.fmut.Lock()
  2282. for deviceID := range fromDevices {
  2283. delete(m.deviceStatRefs, deviceID)
  2284. }
  2285. m.fmut.Unlock()
  2286. scanLimiter.setCapacity(to.Options.MaxConcurrentScans)
  2287. // Some options don't require restart as those components handle it fine
  2288. // by themselves. Compare the options structs containing only the
  2289. // attributes that require restart and act apprioriately.
  2290. if !reflect.DeepEqual(from.Options.RequiresRestartOnly(), to.Options.RequiresRestartOnly()) {
  2291. l.Debugln(m, "requires restart, options differ")
  2292. return false
  2293. }
  2294. return true
  2295. }
  2296. // checkFolderRunningLocked returns nil if the folder is up and running and a
  2297. // descriptive error if not.
  2298. // Need to hold (read) lock on m.fmut when calling this.
  2299. func (m *model) checkFolderRunningLocked(folder string) error {
  2300. _, ok := m.folderRunners[folder]
  2301. if ok {
  2302. return nil
  2303. }
  2304. if cfg, ok := m.cfg.Folder(folder); !ok {
  2305. return errFolderMissing
  2306. } else if cfg.Paused {
  2307. return ErrFolderPaused
  2308. }
  2309. return errFolderNotRunning
  2310. }
  2311. // checkFolderDeviceStatusLocked first checks the folder and then whether the
  2312. // given device is connected and shares this folder.
  2313. // Need to hold (read) lock on both m.fmut and m.pmut when calling this.
  2314. func (m *model) checkDeviceFolderConnectedLocked(device protocol.DeviceID, folder string) error {
  2315. if err := m.checkFolderRunningLocked(folder); err != nil {
  2316. return err
  2317. }
  2318. if cfg, ok := m.cfg.Device(device); !ok {
  2319. return errDeviceUnknown
  2320. } else if cfg.Paused {
  2321. return errDevicePaused
  2322. }
  2323. if _, ok := m.conn[device]; !ok {
  2324. return errors.New("device is not connected")
  2325. }
  2326. if cfg, ok := m.cfg.Folder(folder); !ok || !cfg.SharedWith(device) {
  2327. return errors.New("folder is not shared with device")
  2328. }
  2329. return nil
  2330. }
  2331. // mapFolders returns a map of folder ID to folder configuration for the given
  2332. // slice of folder configurations.
  2333. func mapFolders(folders []config.FolderConfiguration) map[string]config.FolderConfiguration {
  2334. m := make(map[string]config.FolderConfiguration, len(folders))
  2335. for _, cfg := range folders {
  2336. m[cfg.ID] = cfg
  2337. }
  2338. return m
  2339. }
  2340. // mapDevices returns a map of device ID to nothing for the given slice of
  2341. // device IDs.
  2342. func mapDevices(devices []protocol.DeviceID) map[protocol.DeviceID]struct{} {
  2343. m := make(map[protocol.DeviceID]struct{}, len(devices))
  2344. for _, dev := range devices {
  2345. m[dev] = struct{}{}
  2346. }
  2347. return m
  2348. }
  2349. func readOffsetIntoBuf(fs fs.Filesystem, file string, offset int64, buf []byte) error {
  2350. fd, err := fs.Open(file)
  2351. if err != nil {
  2352. l.Debugln("readOffsetIntoBuf.Open", file, err)
  2353. return err
  2354. }
  2355. defer fd.Close()
  2356. _, err = fd.ReadAt(buf, offset)
  2357. if err != nil {
  2358. l.Debugln("readOffsetIntoBuf.ReadAt", file, err)
  2359. }
  2360. return err
  2361. }
  2362. // makeForgetUpdate takes an index update and constructs a download progress update
  2363. // causing to forget any progress for files which we've just been sent.
  2364. func makeForgetUpdate(files []protocol.FileInfo) []protocol.FileDownloadProgressUpdate {
  2365. updates := make([]protocol.FileDownloadProgressUpdate, 0, len(files))
  2366. for _, file := range files {
  2367. if file.IsSymlink() || file.IsDirectory() || file.IsDeleted() {
  2368. continue
  2369. }
  2370. updates = append(updates, protocol.FileDownloadProgressUpdate{
  2371. Name: file.Name,
  2372. Version: file.Version,
  2373. UpdateType: protocol.UpdateTypeForget,
  2374. })
  2375. }
  2376. return updates
  2377. }
  2378. // folderDeviceSet is a set of (folder, deviceID) pairs
  2379. type folderDeviceSet map[string]map[protocol.DeviceID]struct{}
  2380. // set adds the (dev, folder) pair to the set
  2381. func (s folderDeviceSet) set(dev protocol.DeviceID, folder string) {
  2382. devs, ok := s[folder]
  2383. if !ok {
  2384. devs = make(map[protocol.DeviceID]struct{})
  2385. s[folder] = devs
  2386. }
  2387. devs[dev] = struct{}{}
  2388. }
  2389. // has returns true if the (dev, folder) pair is in the set
  2390. func (s folderDeviceSet) has(dev protocol.DeviceID, folder string) bool {
  2391. _, ok := s[folder][dev]
  2392. return ok
  2393. }
  2394. // hasDevice returns true if the device is set on any folder
  2395. func (s folderDeviceSet) hasDevice(dev protocol.DeviceID) bool {
  2396. for _, devices := range s {
  2397. if _, ok := devices[dev]; ok {
  2398. return true
  2399. }
  2400. }
  2401. return false
  2402. }
  2403. type fileInfoBatch struct {
  2404. infos []protocol.FileInfo
  2405. size int
  2406. flushFn func([]protocol.FileInfo) error
  2407. }
  2408. func newFileInfoBatch(fn func([]protocol.FileInfo) error) *fileInfoBatch {
  2409. return &fileInfoBatch{
  2410. infos: make([]protocol.FileInfo, 0, maxBatchSizeFiles),
  2411. flushFn: fn,
  2412. }
  2413. }
  2414. func (b *fileInfoBatch) append(f protocol.FileInfo) {
  2415. b.infos = append(b.infos, f)
  2416. b.size += f.ProtoSize()
  2417. }
  2418. func (b *fileInfoBatch) flushIfFull() error {
  2419. if len(b.infos) >= maxBatchSizeFiles || b.size >= maxBatchSizeBytes {
  2420. return b.flush()
  2421. }
  2422. return nil
  2423. }
  2424. func (b *fileInfoBatch) flush() error {
  2425. if len(b.infos) == 0 {
  2426. return nil
  2427. }
  2428. if err := b.flushFn(b.infos); err != nil {
  2429. return err
  2430. }
  2431. b.reset()
  2432. return nil
  2433. }
  2434. func (b *fileInfoBatch) reset() {
  2435. b.infos = b.infos[:0]
  2436. b.size = 0
  2437. }
  2438. // syncMutexMap is a type safe wrapper for a sync.Map that holds mutexes
  2439. type syncMutexMap struct {
  2440. inner stdsync.Map
  2441. }
  2442. func (m *syncMutexMap) Get(key string) sync.Mutex {
  2443. v, _ := m.inner.LoadOrStore(key, sync.NewMutex())
  2444. return v.(sync.Mutex)
  2445. }
  2446. // sanitizePath takes a string that might contain all kinds of special
  2447. // characters and makes a valid, similar, path name out of it.
  2448. //
  2449. // Spans of invalid characters are replaced by a single space. Invalid
  2450. // characters are control characters, the things not allowed in file names
  2451. // in Windows, and common shell metacharacters. Even if asterisks and pipes
  2452. // and stuff are allowed on Unixes in general they might not be allowed by
  2453. // the filesystem and may surprise the user and cause shell oddness. This
  2454. // function is intended for file names we generate on behalf of the user,
  2455. // and surprising them with odd shell characters in file names is unkind.
  2456. //
  2457. // We include whitespace in the invalid characters so that multiple
  2458. // whitespace is collapsed to a single space. Additionally, whitespace at
  2459. // either end is removed.
  2460. func sanitizePath(path string) string {
  2461. invalid := regexp.MustCompile(`([[:cntrl:]]|[<>:"'/\\|?*\n\r\t \[\]\{\};:!@$%&^#])+`)
  2462. return strings.TrimSpace(invalid.ReplaceAllString(path, " "))
  2463. }