folder_sendrecv_test.go 40 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407
  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. "crypto/rand"
  11. "errors"
  12. "fmt"
  13. "io"
  14. "os"
  15. "path/filepath"
  16. "strconv"
  17. "strings"
  18. "testing"
  19. "time"
  20. "github.com/syncthing/syncthing/lib/build"
  21. "github.com/syncthing/syncthing/lib/config"
  22. "github.com/syncthing/syncthing/lib/events"
  23. "github.com/syncthing/syncthing/lib/fs"
  24. "github.com/syncthing/syncthing/lib/ignore"
  25. "github.com/syncthing/syncthing/lib/osutil"
  26. "github.com/syncthing/syncthing/lib/protocol"
  27. "github.com/syncthing/syncthing/lib/scanner"
  28. "github.com/syncthing/syncthing/lib/sync"
  29. )
  30. var blocks = []protocol.BlockInfo{
  31. {Hash: []uint8{0xfa, 0x43, 0x23, 0x9b, 0xce, 0xe7, 0xb9, 0x7c, 0xa6, 0x2f, 0x0, 0x7c, 0xc6, 0x84, 0x87, 0x56, 0xa, 0x39, 0xe1, 0x9f, 0x74, 0xf3, 0xdd, 0xe7, 0x48, 0x6d, 0xb3, 0xf9, 0x8d, 0xf8, 0xe4, 0x71}}, // Zero'ed out block
  32. {Offset: 0, Size: 0x20000, Hash: []uint8{0x7e, 0xad, 0xbc, 0x36, 0xae, 0xbb, 0xcf, 0x74, 0x43, 0xe2, 0x7a, 0x5a, 0x4b, 0xb8, 0x5b, 0xce, 0xe6, 0x9e, 0x1e, 0x10, 0xf9, 0x8a, 0xbc, 0x77, 0x95, 0x2, 0x29, 0x60, 0x9e, 0x96, 0xae, 0x6c}},
  33. {Offset: 131072, Size: 0x20000, Hash: []uint8{0x3c, 0xc4, 0x20, 0xf4, 0xb, 0x2e, 0xcb, 0xb9, 0x5d, 0xce, 0x34, 0xa8, 0xc3, 0x92, 0xea, 0xf3, 0xda, 0x88, 0x33, 0xee, 0x7a, 0xb6, 0xe, 0xf1, 0x82, 0x5e, 0xb0, 0xa9, 0x26, 0xa9, 0xc0, 0xef}},
  34. {Offset: 262144, Size: 0x20000, Hash: []uint8{0x76, 0xa8, 0xc, 0x69, 0xd7, 0x5c, 0x52, 0xfd, 0xdf, 0x55, 0xef, 0x44, 0xc1, 0xd6, 0x25, 0x48, 0x4d, 0x98, 0x48, 0x4d, 0xaa, 0x50, 0xf6, 0x6b, 0x32, 0x47, 0x55, 0x81, 0x6b, 0xed, 0xee, 0xfb}},
  35. {Offset: 393216, Size: 0x20000, Hash: []uint8{0x44, 0x1e, 0xa4, 0xf2, 0x8d, 0x1f, 0xc3, 0x1b, 0x9d, 0xa5, 0x18, 0x5e, 0x59, 0x1b, 0xd8, 0x5c, 0xba, 0x7d, 0xb9, 0x8d, 0x70, 0x11, 0x5c, 0xea, 0xa1, 0x57, 0x4d, 0xcb, 0x3c, 0x5b, 0xf8, 0x6c}},
  36. {Offset: 524288, Size: 0x20000, Hash: []uint8{0x8, 0x40, 0xd0, 0x5e, 0x80, 0x0, 0x0, 0x7c, 0x8b, 0xb3, 0x8b, 0xf7, 0x7b, 0x23, 0x26, 0x28, 0xab, 0xda, 0xcf, 0x86, 0x8f, 0xc2, 0x8a, 0x39, 0xc6, 0xe6, 0x69, 0x59, 0x97, 0xb6, 0x1a, 0x43}},
  37. {Offset: 655360, Size: 0x20000, Hash: []uint8{0x38, 0x8e, 0x44, 0xcb, 0x30, 0xd8, 0x90, 0xf, 0xce, 0x7, 0x4b, 0x58, 0x86, 0xde, 0xce, 0x59, 0xa2, 0x46, 0xd2, 0xf9, 0xba, 0xaf, 0x35, 0x87, 0x38, 0xdf, 0xd2, 0xd, 0xf9, 0x45, 0xed, 0x91}},
  38. {Offset: 786432, Size: 0x20000, Hash: []uint8{0x32, 0x28, 0xcd, 0xf, 0x37, 0x21, 0xe5, 0xd4, 0x1e, 0x58, 0x87, 0x73, 0x8e, 0x36, 0xdf, 0xb2, 0x70, 0x78, 0x56, 0xc3, 0x42, 0xff, 0xf7, 0x8f, 0x37, 0x95, 0x0, 0x26, 0xa, 0xac, 0x54, 0x72}},
  39. {Offset: 917504, Size: 0x20000, Hash: []uint8{0x96, 0x6b, 0x15, 0x6b, 0xc4, 0xf, 0x19, 0x18, 0xca, 0xbb, 0x5f, 0xd6, 0xbb, 0xa2, 0xc6, 0x2a, 0xac, 0xbb, 0x8a, 0xb9, 0xce, 0xec, 0x4c, 0xdb, 0x78, 0xec, 0x57, 0x5d, 0x33, 0xf9, 0x8e, 0xaf}},
  40. }
  41. var folders = []string{"default"}
  42. var diffTestData = []struct {
  43. a string
  44. b string
  45. s int
  46. d []protocol.BlockInfo
  47. }{
  48. {"contents", "contents", 1024, []protocol.BlockInfo{}},
  49. {"", "", 1024, []protocol.BlockInfo{}},
  50. {"contents", "contents", 3, []protocol.BlockInfo{}},
  51. {"contents", "cantents", 3, []protocol.BlockInfo{{Offset: 0, Size: 3}}},
  52. {"contents", "contants", 3, []protocol.BlockInfo{{Offset: 3, Size: 3}}},
  53. {"contents", "cantants", 3, []protocol.BlockInfo{{Offset: 0, Size: 3}, {Offset: 3, Size: 3}}},
  54. {"contents", "", 3, []protocol.BlockInfo{{Offset: 0, Size: 0}}},
  55. {"", "contents", 3, []protocol.BlockInfo{{Offset: 0, Size: 3}, {Offset: 3, Size: 3}, {Offset: 6, Size: 2}}},
  56. {"con", "contents", 3, []protocol.BlockInfo{{Offset: 3, Size: 3}, {Offset: 6, Size: 2}}},
  57. {"contents", "con", 3, nil},
  58. {"contents", "cont", 3, []protocol.BlockInfo{{Offset: 3, Size: 1}}},
  59. {"cont", "contents", 3, []protocol.BlockInfo{{Offset: 3, Size: 3}, {Offset: 6, Size: 2}}},
  60. }
  61. func setupFile(filename string, blockNumbers []int) protocol.FileInfo {
  62. // Create existing file
  63. existingBlocks := make([]protocol.BlockInfo, len(blockNumbers))
  64. for i := range blockNumbers {
  65. existingBlocks[i] = blocks[blockNumbers[i]]
  66. }
  67. return protocol.FileInfo{
  68. Name: filename,
  69. Blocks: existingBlocks,
  70. }
  71. }
  72. func createEmptyFileInfo(t *testing.T, name string, fs fs.Filesystem) protocol.FileInfo {
  73. t.Helper()
  74. writeFile(t, fs, name, nil)
  75. fi, err := fs.Stat(name)
  76. must(t, err)
  77. file, err := scanner.CreateFileInfo(fi, name, fs, false, false, config.XattrFilter{})
  78. must(t, err)
  79. return file
  80. }
  81. // Sets up a folder and model, but makes sure the services aren't actually running.
  82. func setupSendReceiveFolder(t testing.TB, files ...protocol.FileInfo) (*testModel, *sendReceiveFolder, context.CancelFunc) {
  83. w, fcfg, wCancel := tmpDefaultWrapper(t)
  84. // Initialise model and stop immediately.
  85. model := setupModel(t, w)
  86. model.cancel()
  87. <-model.stopped
  88. f := model.folderRunners[fcfg.ID].(*sendReceiveFolder)
  89. f.tempPullErrors = make(map[string]string)
  90. f.ctx = context.Background()
  91. // Update index
  92. if files != nil {
  93. f.updateLocalsFromScanning(files)
  94. }
  95. return model, f, wCancel
  96. }
  97. func cleanupSRFolder(f *sendReceiveFolder, m *testModel, wrapperCancel context.CancelFunc) {
  98. wrapperCancel()
  99. os.Remove(m.cfg.ConfigPath())
  100. os.RemoveAll(f.Filesystem(nil).URI())
  101. }
  102. // Layout of the files: (indexes from the above array)
  103. // 12345678 - Required file
  104. // 02005008 - Existing file (currently in the index)
  105. // 02340070 - Temp file on the disk
  106. func TestHandleFile(t *testing.T) {
  107. // After the diff between required and existing we should:
  108. // Copy: 2, 5, 8
  109. // Pull: 1, 3, 4, 6, 7
  110. existingBlocks := []int{0, 2, 0, 0, 5, 0, 0, 8}
  111. existingFile := setupFile("filex", existingBlocks)
  112. requiredFile := existingFile
  113. requiredFile.Blocks = blocks[1:]
  114. m, f, wcfgCancel := setupSendReceiveFolder(t, existingFile)
  115. defer cleanupSRFolder(f, m, wcfgCancel)
  116. copyChan := make(chan copyBlocksState, 1)
  117. f.handleFile(requiredFile, fsetSnapshot(t, f.fset), copyChan)
  118. // Receive the results
  119. toCopy := <-copyChan
  120. if len(toCopy.blocks) != 8 {
  121. t.Errorf("Unexpected count of copy blocks: %d != 8", len(toCopy.blocks))
  122. }
  123. for _, block := range blocks[1:] {
  124. found := false
  125. for _, toCopyBlock := range toCopy.blocks {
  126. if bytes.Equal(toCopyBlock.Hash, block.Hash) {
  127. found = true
  128. break
  129. }
  130. }
  131. if !found {
  132. t.Errorf("Did not find block %s", block.String())
  133. }
  134. }
  135. }
  136. func TestHandleFileWithTemp(t *testing.T) {
  137. // After diff between required and existing we should:
  138. // Copy: 2, 5, 8
  139. // Pull: 1, 3, 4, 6, 7
  140. // After dropping out blocks already on the temp file we should:
  141. // Copy: 5, 8
  142. // Pull: 1, 6
  143. existingBlocks := []int{0, 2, 0, 0, 5, 0, 0, 8}
  144. existingFile := setupFile("file", existingBlocks)
  145. requiredFile := existingFile
  146. requiredFile.Blocks = blocks[1:]
  147. m, f, wcfgCancel := setupSendReceiveFolder(t, existingFile)
  148. defer cleanupSRFolder(f, m, wcfgCancel)
  149. if _, err := prepareTmpFile(f.Filesystem(nil)); err != nil {
  150. t.Fatal(err)
  151. }
  152. copyChan := make(chan copyBlocksState, 1)
  153. f.handleFile(requiredFile, fsetSnapshot(t, f.fset), copyChan)
  154. // Receive the results
  155. toCopy := <-copyChan
  156. if len(toCopy.blocks) != 4 {
  157. t.Errorf("Unexpected count of copy blocks: %d != 4", len(toCopy.blocks))
  158. }
  159. for _, idx := range []int{1, 5, 6, 8} {
  160. found := false
  161. block := blocks[idx]
  162. for _, toCopyBlock := range toCopy.blocks {
  163. if bytes.Equal(toCopyBlock.Hash, block.Hash) {
  164. found = true
  165. break
  166. }
  167. }
  168. if !found {
  169. t.Errorf("Did not find block %s", block.String())
  170. }
  171. }
  172. }
  173. func TestCopierFinder(t *testing.T) {
  174. methods := []fs.CopyRangeMethod{fs.CopyRangeMethodStandard, fs.CopyRangeMethodAllWithFallback}
  175. if build.IsLinux {
  176. methods = append(methods, fs.CopyRangeMethodSendFile)
  177. }
  178. for _, method := range methods {
  179. t.Run(method.String(), func(t *testing.T) {
  180. // After diff between required and existing we should:
  181. // Copy: 1, 2, 3, 4, 6, 7, 8
  182. // Since there is no existing file, nor a temp file
  183. // After dropping out blocks found locally:
  184. // Pull: 1, 5, 6, 8
  185. tempFile := fs.TempName("file2")
  186. existingBlocks := []int{0, 2, 3, 4, 0, 0, 7, 0}
  187. existingFile := setupFile(fs.TempName("file"), existingBlocks)
  188. existingFile.Size = 1
  189. requiredFile := existingFile
  190. requiredFile.Blocks = blocks[1:]
  191. requiredFile.Name = "file2"
  192. m, f, wcfgCancel := setupSendReceiveFolder(t, existingFile)
  193. f.CopyRangeMethod = method
  194. defer cleanupSRFolder(f, m, wcfgCancel)
  195. if _, err := prepareTmpFile(f.Filesystem(nil)); err != nil {
  196. t.Fatal(err)
  197. }
  198. copyChan := make(chan copyBlocksState)
  199. pullChan := make(chan pullBlockState, 4)
  200. finisherChan := make(chan *sharedPullerState, 1)
  201. // Run a single fetcher routine
  202. go f.copierRoutine(copyChan, pullChan, finisherChan)
  203. defer close(copyChan)
  204. f.handleFile(requiredFile, fsetSnapshot(t, f.fset), copyChan)
  205. timeout := time.After(10 * time.Second)
  206. pulls := make([]pullBlockState, 4)
  207. for i := 0; i < 4; i++ {
  208. select {
  209. case pulls[i] = <-pullChan:
  210. case <-timeout:
  211. t.Fatalf("Timed out before receiving all 4 states on pullChan (already got %v)", i)
  212. }
  213. }
  214. var finish *sharedPullerState
  215. select {
  216. case finish = <-finisherChan:
  217. case <-timeout:
  218. t.Fatal("Timed out before receiving 4 states on pullChan")
  219. }
  220. defer cleanupSharedPullerState(finish)
  221. select {
  222. case <-pullChan:
  223. t.Fatal("Pull channel has data to be read")
  224. case <-finisherChan:
  225. t.Fatal("Finisher channel has data to be read")
  226. default:
  227. }
  228. // Verify that the right blocks went into the pull list.
  229. // They are pulled in random order.
  230. for _, idx := range []int{1, 5, 6, 8} {
  231. found := false
  232. block := blocks[idx]
  233. for _, pulledBlock := range pulls {
  234. if bytes.Equal(pulledBlock.block.Hash, block.Hash) {
  235. found = true
  236. break
  237. }
  238. }
  239. if !found {
  240. t.Errorf("Did not find block %s", block.String())
  241. }
  242. if !bytes.Equal(finish.file.Blocks[idx-1].Hash, blocks[idx].Hash) {
  243. t.Errorf("Block %d mismatch: %s != %s", idx, finish.file.Blocks[idx-1].String(), blocks[idx].String())
  244. }
  245. }
  246. // Verify that the fetched blocks have actually been written to the temp file
  247. blks, err := scanner.HashFile(context.TODO(), f.Filesystem(nil), tempFile, protocol.MinBlockSize, nil, false)
  248. if err != nil {
  249. t.Log(err)
  250. }
  251. for _, eq := range []int{2, 3, 4, 7} {
  252. if !bytes.Equal(blks[eq-1].Hash, blocks[eq].Hash) {
  253. t.Errorf("Block %d mismatch: %s != %s", eq, blks[eq-1].String(), blocks[eq].String())
  254. }
  255. }
  256. })
  257. }
  258. }
  259. func TestWeakHash(t *testing.T) {
  260. // Setup the model/pull environment
  261. model, fo, wcfgCancel := setupSendReceiveFolder(t)
  262. defer cleanupSRFolder(fo, model, wcfgCancel)
  263. ffs := fo.Filesystem(nil)
  264. tempFile := fs.TempName("weakhash")
  265. var shift int64 = 10
  266. var size int64 = 1 << 20
  267. expectBlocks := int(size / protocol.MinBlockSize)
  268. expectPulls := int(shift / protocol.MinBlockSize)
  269. if shift > 0 {
  270. expectPulls++
  271. }
  272. f, err := ffs.Create("weakhash")
  273. must(t, err)
  274. defer f.Close()
  275. _, err = io.CopyN(f, rand.Reader, size)
  276. if err != nil {
  277. t.Error(err)
  278. }
  279. info, err := f.Stat()
  280. if err != nil {
  281. t.Error(err)
  282. }
  283. // Create two files, second file has `shifted` bytes random prefix, yet
  284. // both are of the same length, for example:
  285. // File 1: abcdefgh
  286. // File 2: xyabcdef
  287. f.Seek(0, io.SeekStart)
  288. existing, err := scanner.Blocks(context.TODO(), f, protocol.MinBlockSize, size, nil, true)
  289. if err != nil {
  290. t.Error(err)
  291. }
  292. f.Seek(0, io.SeekStart)
  293. remainder := io.LimitReader(f, size-shift)
  294. prefix := io.LimitReader(rand.Reader, shift)
  295. nf := io.MultiReader(prefix, remainder)
  296. desired, err := scanner.Blocks(context.TODO(), nf, protocol.MinBlockSize, size, nil, true)
  297. if err != nil {
  298. t.Error(err)
  299. }
  300. existingFile := protocol.FileInfo{
  301. Name: "weakhash",
  302. Blocks: existing,
  303. Size: size,
  304. ModifiedS: info.ModTime().Unix(),
  305. ModifiedNs: info.ModTime().Nanosecond(),
  306. }
  307. desiredFile := protocol.FileInfo{
  308. Name: "weakhash",
  309. Size: size,
  310. Blocks: desired,
  311. ModifiedS: info.ModTime().Unix() + 1,
  312. }
  313. fo.updateLocalsFromScanning([]protocol.FileInfo{existingFile})
  314. copyChan := make(chan copyBlocksState)
  315. pullChan := make(chan pullBlockState, expectBlocks)
  316. finisherChan := make(chan *sharedPullerState, 1)
  317. // Run a single fetcher routine
  318. go fo.copierRoutine(copyChan, pullChan, finisherChan)
  319. defer close(copyChan)
  320. // Test 1 - no weak hashing, file gets fully repulled (`expectBlocks` pulls).
  321. fo.WeakHashThresholdPct = 101
  322. fo.handleFile(desiredFile, fsetSnapshot(t, fo.fset), copyChan)
  323. var pulls []pullBlockState
  324. timeout := time.After(10 * time.Second)
  325. for len(pulls) < expectBlocks {
  326. select {
  327. case pull := <-pullChan:
  328. pulls = append(pulls, pull)
  329. case <-timeout:
  330. t.Fatalf("timed out, got %d pulls expected %d", len(pulls), expectPulls)
  331. }
  332. }
  333. finish := <-finisherChan
  334. select {
  335. case <-pullChan:
  336. t.Fatal("Pull channel has data to be read")
  337. case <-finisherChan:
  338. t.Fatal("Finisher channel has data to be read")
  339. default:
  340. }
  341. cleanupSharedPullerState(finish)
  342. if err := ffs.Remove(tempFile); err != nil {
  343. t.Fatal(err)
  344. }
  345. // Test 2 - using weak hash, expectPulls blocks pulled.
  346. fo.WeakHashThresholdPct = -1
  347. fo.handleFile(desiredFile, fsetSnapshot(t, fo.fset), copyChan)
  348. pulls = pulls[:0]
  349. for len(pulls) < expectPulls {
  350. select {
  351. case pull := <-pullChan:
  352. pulls = append(pulls, pull)
  353. case <-time.After(10 * time.Second):
  354. t.Fatalf("timed out, got %d pulls expected %d", len(pulls), expectPulls)
  355. }
  356. }
  357. finish = <-finisherChan
  358. cleanupSharedPullerState(finish)
  359. expectShifted := expectBlocks - expectPulls
  360. if finish.copyOriginShifted != expectShifted {
  361. t.Errorf("did not copy %d shifted", expectShifted)
  362. }
  363. }
  364. // Test that updating a file removes its old blocks from the blockmap
  365. func TestCopierCleanup(t *testing.T) {
  366. iterFn := func(folder, file string, index int32) bool {
  367. return true
  368. }
  369. // Create a file
  370. file := setupFile("test", []int{0})
  371. file.Size = 1
  372. m, f, wcfgCancel := setupSendReceiveFolder(t, file)
  373. defer cleanupSRFolder(f, m, wcfgCancel)
  374. file.Blocks = []protocol.BlockInfo{blocks[1]}
  375. file.Version = file.Version.Update(myID.Short())
  376. // Update index (removing old blocks)
  377. f.updateLocalsFromScanning([]protocol.FileInfo{file})
  378. if m.finder.Iterate(folders, blocks[0].Hash, iterFn) {
  379. t.Error("Unexpected block found")
  380. }
  381. if !m.finder.Iterate(folders, blocks[1].Hash, iterFn) {
  382. t.Error("Expected block not found")
  383. }
  384. file.Blocks = []protocol.BlockInfo{blocks[0]}
  385. file.Version = file.Version.Update(myID.Short())
  386. // Update index (removing old blocks)
  387. f.updateLocalsFromScanning([]protocol.FileInfo{file})
  388. if !m.finder.Iterate(folders, blocks[0].Hash, iterFn) {
  389. t.Error("Unexpected block found")
  390. }
  391. if m.finder.Iterate(folders, blocks[1].Hash, iterFn) {
  392. t.Error("Expected block not found")
  393. }
  394. }
  395. func TestDeregisterOnFailInCopy(t *testing.T) {
  396. file := setupFile("filex", []int{0, 2, 0, 0, 5, 0, 0, 8})
  397. m, f, wcfgCancel := setupSendReceiveFolder(t)
  398. defer cleanupSRFolder(f, m, wcfgCancel)
  399. // Set up our evet subscription early
  400. s := m.evLogger.Subscribe(events.ItemFinished)
  401. // queue.Done should be called by the finisher routine
  402. f.queue.Push("filex", 0, time.Time{})
  403. f.queue.Pop()
  404. if f.queue.lenProgress() != 1 {
  405. t.Fatal("Expected file in progress")
  406. }
  407. pullChan := make(chan pullBlockState)
  408. finisherBufferChan := make(chan *sharedPullerState, 1)
  409. finisherChan := make(chan *sharedPullerState)
  410. dbUpdateChan := make(chan dbUpdateJob, 1)
  411. snap := fsetSnapshot(t, f.fset)
  412. copyChan, copyWg := startCopier(f, pullChan, finisherBufferChan)
  413. go f.finisherRoutine(snap, finisherChan, dbUpdateChan, make(chan string))
  414. defer func() {
  415. close(copyChan)
  416. copyWg.Wait()
  417. close(pullChan)
  418. close(finisherBufferChan)
  419. close(finisherChan)
  420. }()
  421. f.handleFile(file, snap, copyChan)
  422. // Receive a block at puller, to indicate that at least a single copier
  423. // loop has been performed.
  424. var toPull pullBlockState
  425. select {
  426. case toPull = <-pullChan:
  427. case <-time.After(10 * time.Second):
  428. t.Fatal("timed out")
  429. }
  430. // Unblock copier
  431. go func() {
  432. for range pullChan {
  433. }
  434. }()
  435. // Close the file, causing errors on further access
  436. toPull.sharedPullerState.fail(os.ErrNotExist)
  437. select {
  438. case state := <-finisherBufferChan:
  439. // At this point the file should still be registered with both the job
  440. // queue, and the progress emitter. Verify this.
  441. if f.model.progressEmitter.lenRegistry() != 1 || f.queue.lenProgress() != 1 || f.queue.lenQueued() != 0 {
  442. t.Fatal("Could not find file")
  443. }
  444. // Pass the file down the real finisher, and give it time to consume
  445. finisherChan <- state
  446. t0 := time.Now()
  447. if ev, err := s.Poll(time.Minute); err != nil {
  448. t.Fatal("Got error waiting for ItemFinished event:", err)
  449. } else if n := ev.Data.(map[string]interface{})["item"]; n != state.file.Name {
  450. t.Fatal("Got ItemFinished event for wrong file:", n)
  451. }
  452. t.Log("event took", time.Since(t0))
  453. state.mut.Lock()
  454. stateWriter := state.writer
  455. state.mut.Unlock()
  456. if stateWriter != nil {
  457. t.Fatal("File not closed?")
  458. }
  459. if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
  460. t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
  461. }
  462. // Doing it again should have no effect
  463. finisherChan <- state
  464. if _, err := s.Poll(time.Second); err != events.ErrTimeout {
  465. t.Fatal("Expected timeout, not another event", err)
  466. }
  467. if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
  468. t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
  469. }
  470. case <-time.After(5 * time.Second):
  471. t.Fatal("Didn't get anything to the finisher")
  472. }
  473. }
  474. func TestDeregisterOnFailInPull(t *testing.T) {
  475. file := setupFile("filex", []int{0, 2, 0, 0, 5, 0, 0, 8})
  476. m, f, wcfgCancel := setupSendReceiveFolder(t)
  477. defer cleanupSRFolder(f, m, wcfgCancel)
  478. // Set up our evet subscription early
  479. s := m.evLogger.Subscribe(events.ItemFinished)
  480. // queue.Done should be called by the finisher routine
  481. f.queue.Push("filex", 0, time.Time{})
  482. f.queue.Pop()
  483. if f.queue.lenProgress() != 1 {
  484. t.Fatal("Expected file in progress")
  485. }
  486. pullChan := make(chan pullBlockState)
  487. finisherBufferChan := make(chan *sharedPullerState)
  488. finisherChan := make(chan *sharedPullerState)
  489. dbUpdateChan := make(chan dbUpdateJob, 1)
  490. snap := fsetSnapshot(t, f.fset)
  491. copyChan, copyWg := startCopier(f, pullChan, finisherBufferChan)
  492. pullWg := sync.NewWaitGroup()
  493. pullWg.Add(1)
  494. go func() {
  495. f.pullerRoutine(snap, pullChan, finisherBufferChan)
  496. pullWg.Done()
  497. }()
  498. go f.finisherRoutine(snap, finisherChan, dbUpdateChan, make(chan string))
  499. defer func() {
  500. // Unblock copier and puller
  501. go func() {
  502. for range finisherBufferChan {
  503. }
  504. }()
  505. close(copyChan)
  506. copyWg.Wait()
  507. close(pullChan)
  508. pullWg.Wait()
  509. close(finisherBufferChan)
  510. close(finisherChan)
  511. }()
  512. f.handleFile(file, snap, copyChan)
  513. // Receive at finisher, we should error out as puller has nowhere to pull
  514. // from.
  515. timeout = time.Second
  516. // Both the puller and copier may send to the finisherBufferChan.
  517. var state *sharedPullerState
  518. after := time.After(5 * time.Second)
  519. for {
  520. select {
  521. case state = <-finisherBufferChan:
  522. case <-after:
  523. t.Fatal("Didn't get failed state to the finisher")
  524. }
  525. if state.failed() != nil {
  526. break
  527. }
  528. }
  529. // At this point the file should still be registered with both the job
  530. // queue, and the progress emitter. Verify this.
  531. if f.model.progressEmitter.lenRegistry() != 1 || f.queue.lenProgress() != 1 || f.queue.lenQueued() != 0 {
  532. t.Fatal("Could not find file")
  533. }
  534. // Pass the file down the real finisher, and give it time to consume
  535. finisherChan <- state
  536. t0 := time.Now()
  537. if ev, err := s.Poll(time.Minute); err != nil {
  538. t.Fatal("Got error waiting for ItemFinished event:", err)
  539. } else if n := ev.Data.(map[string]interface{})["item"]; n != state.file.Name {
  540. t.Fatal("Got ItemFinished event for wrong file:", n)
  541. }
  542. t.Log("event took", time.Since(t0))
  543. state.mut.Lock()
  544. stateWriter := state.writer
  545. state.mut.Unlock()
  546. if stateWriter != nil {
  547. t.Fatal("File not closed?")
  548. }
  549. if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
  550. t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
  551. }
  552. // Doing it again should have no effect
  553. finisherChan <- state
  554. if _, err := s.Poll(time.Second); err != events.ErrTimeout {
  555. t.Fatal("Expected timeout, not another event", err)
  556. }
  557. if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
  558. t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
  559. }
  560. }
  561. func TestIssue3164(t *testing.T) {
  562. m, f, wcfgCancel := setupSendReceiveFolder(t)
  563. defer cleanupSRFolder(f, m, wcfgCancel)
  564. ffs := f.Filesystem(nil)
  565. tmpDir := ffs.URI()
  566. ignDir := filepath.Join("issue3164", "oktodelete")
  567. subDir := filepath.Join(ignDir, "foobar")
  568. must(t, ffs.MkdirAll(subDir, 0777))
  569. must(t, os.WriteFile(filepath.Join(tmpDir, subDir, "file"), []byte("Hello"), 0644))
  570. must(t, os.WriteFile(filepath.Join(tmpDir, ignDir, "file"), []byte("Hello"), 0644))
  571. file := protocol.FileInfo{
  572. Name: "issue3164",
  573. }
  574. must(t, f.scanSubdirs(nil))
  575. matcher := ignore.New(ffs)
  576. must(t, matcher.Parse(bytes.NewBufferString("(?d)oktodelete"), ""))
  577. f.ignores = matcher
  578. dbUpdateChan := make(chan dbUpdateJob, 1)
  579. f.deleteDir(file, fsetSnapshot(t, f.fset), dbUpdateChan, make(chan string))
  580. if _, err := ffs.Stat("issue3164"); !fs.IsNotExist(err) {
  581. t.Fatal(err)
  582. }
  583. }
  584. func TestDiff(t *testing.T) {
  585. for i, test := range diffTestData {
  586. a, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.a), test.s, -1, nil, false)
  587. b, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.b), test.s, -1, nil, false)
  588. _, d := blockDiff(a, b)
  589. if len(d) != len(test.d) {
  590. t.Fatalf("Incorrect length for diff %d; %d != %d", i, len(d), len(test.d))
  591. } else {
  592. for j := range test.d {
  593. if d[j].Offset != test.d[j].Offset {
  594. t.Errorf("Incorrect offset for diff %d block %d; %d != %d", i, j, d[j].Offset, test.d[j].Offset)
  595. }
  596. if d[j].Size != test.d[j].Size {
  597. t.Errorf("Incorrect length for diff %d block %d; %d != %d", i, j, d[j].Size, test.d[j].Size)
  598. }
  599. }
  600. }
  601. }
  602. }
  603. func BenchmarkDiff(b *testing.B) {
  604. testCases := make([]struct{ a, b []protocol.BlockInfo }, 0, len(diffTestData))
  605. for _, test := range diffTestData {
  606. a, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.a), test.s, -1, nil, false)
  607. b, _ := scanner.Blocks(context.TODO(), bytes.NewBufferString(test.b), test.s, -1, nil, false)
  608. testCases = append(testCases, struct{ a, b []protocol.BlockInfo }{a, b})
  609. }
  610. b.ReportAllocs()
  611. b.ResetTimer()
  612. for i := 0; i < b.N; i++ {
  613. for _, tc := range testCases {
  614. blockDiff(tc.a, tc.b)
  615. }
  616. }
  617. }
  618. func TestDiffEmpty(t *testing.T) {
  619. emptyCases := []struct {
  620. a []protocol.BlockInfo
  621. b []protocol.BlockInfo
  622. need int
  623. have int
  624. }{
  625. {nil, nil, 0, 0},
  626. {[]protocol.BlockInfo{{Offset: 3, Size: 1}}, nil, 0, 0},
  627. {nil, []protocol.BlockInfo{{Offset: 3, Size: 1}}, 1, 0},
  628. }
  629. for _, emptyCase := range emptyCases {
  630. h, n := blockDiff(emptyCase.a, emptyCase.b)
  631. if len(h) != emptyCase.have {
  632. t.Errorf("incorrect have: %d != %d", len(h), emptyCase.have)
  633. }
  634. if len(n) != emptyCase.need {
  635. t.Errorf("incorrect have: %d != %d", len(h), emptyCase.have)
  636. }
  637. }
  638. }
  639. // TestDeleteIgnorePerms checks, that a file gets deleted when the IgnorePerms
  640. // option is true and the permissions do not match between the file on disk and
  641. // in the db.
  642. func TestDeleteIgnorePerms(t *testing.T) {
  643. m, f, wcfgCancel := setupSendReceiveFolder(t)
  644. defer cleanupSRFolder(f, m, wcfgCancel)
  645. ffs := f.Filesystem(nil)
  646. f.IgnorePerms = true
  647. name := "deleteIgnorePerms"
  648. file, err := ffs.Create(name)
  649. if err != nil {
  650. t.Error(err)
  651. }
  652. defer file.Close()
  653. stat, err := file.Stat()
  654. must(t, err)
  655. fi, err := scanner.CreateFileInfo(stat, name, ffs, false, false, config.XattrFilter{})
  656. must(t, err)
  657. ffs.Chmod(name, 0600)
  658. if info, err := ffs.Stat(name); err == nil {
  659. fi.InodeChangeNs = info.InodeChangeTime().UnixNano()
  660. }
  661. scanChan := make(chan string, 1)
  662. err = f.checkToBeDeleted(fi, fi, true, scanChan)
  663. must(t, err)
  664. }
  665. func TestCopyOwner(t *testing.T) {
  666. // Verifies that owner and group are copied from the parent, for both
  667. // files and directories.
  668. if build.IsWindows {
  669. t.Skip("copying owner not supported on Windows")
  670. }
  671. const (
  672. expOwner = 1234
  673. expGroup = 5678
  674. )
  675. // Set up a folder with the CopyParentOwner bit and backed by a fake
  676. // filesystem.
  677. m, f, wcfgCancel := setupSendReceiveFolder(t)
  678. defer cleanupSRFolder(f, m, wcfgCancel)
  679. f.folder.FolderConfiguration = newFolderConfiguration(m.cfg, f.ID, f.Label, fs.FilesystemTypeFake, "/TestCopyOwner")
  680. f.folder.FolderConfiguration.CopyOwnershipFromParent = true
  681. f.fset = newFileSet(t, f.ID, m.db)
  682. f.mtimefs = f.Filesystem(f.fset)
  683. // Create a parent dir with a certain owner/group.
  684. f.mtimefs.Mkdir("foo", 0755)
  685. f.mtimefs.Lchown("foo", strconv.Itoa(expOwner), strconv.Itoa(expGroup))
  686. dir := protocol.FileInfo{
  687. Name: "foo/bar",
  688. Type: protocol.FileInfoTypeDirectory,
  689. Permissions: 0755,
  690. }
  691. // Have the folder create a subdirectory, verify that it's the correct
  692. // owner/group.
  693. dbUpdateChan := make(chan dbUpdateJob, 1)
  694. scanChan := make(chan string)
  695. defer close(dbUpdateChan)
  696. f.handleDir(dir, fsetSnapshot(t, f.fset), dbUpdateChan, scanChan)
  697. select {
  698. case <-dbUpdateChan: // empty the channel for later
  699. case toScan := <-scanChan:
  700. t.Fatal("Unexpected receive on scanChan:", toScan)
  701. }
  702. info, err := f.mtimefs.Lstat("foo/bar")
  703. if err != nil {
  704. t.Fatal("Unexpected error (dir):", err)
  705. }
  706. if info.Owner() != expOwner || info.Group() != expGroup {
  707. t.Fatalf("Expected dir owner/group to be %d/%d, not %d/%d", expOwner, expGroup, info.Owner(), info.Group())
  708. }
  709. // Have the folder create a file, verify it's the correct owner/group.
  710. // File is zero sized to avoid having to handle copies/pulls.
  711. file := protocol.FileInfo{
  712. Name: "foo/bar/baz",
  713. Type: protocol.FileInfoTypeFile,
  714. Permissions: 0644,
  715. }
  716. // Wire some stuff. The flow here is handleFile() -[copierChan]->
  717. // copierRoutine() -[finisherChan]-> finisherRoutine() -[dbUpdateChan]->
  718. // back to us and we're done. The copier routine doesn't do anything,
  719. // but it's the way data is passed around. When the database update
  720. // comes the finisher is done.
  721. snap := fsetSnapshot(t, f.fset)
  722. finisherChan := make(chan *sharedPullerState)
  723. copierChan, copyWg := startCopier(f, nil, finisherChan)
  724. go f.finisherRoutine(snap, finisherChan, dbUpdateChan, nil)
  725. defer func() {
  726. close(copierChan)
  727. copyWg.Wait()
  728. close(finisherChan)
  729. }()
  730. f.handleFile(file, snap, copierChan)
  731. <-dbUpdateChan
  732. info, err = f.mtimefs.Lstat("foo/bar/baz")
  733. if err != nil {
  734. t.Fatal("Unexpected error (file):", err)
  735. }
  736. if info.Owner() != expOwner || info.Group() != expGroup {
  737. t.Fatalf("Expected file owner/group to be %d/%d, not %d/%d", expOwner, expGroup, info.Owner(), info.Group())
  738. }
  739. // Have the folder create a symlink. Verify it accordingly.
  740. symlink := protocol.FileInfo{
  741. Name: "foo/bar/sym",
  742. Type: protocol.FileInfoTypeSymlink,
  743. Permissions: 0644,
  744. SymlinkTarget: "over the rainbow",
  745. }
  746. f.handleSymlink(symlink, snap, dbUpdateChan, scanChan)
  747. select {
  748. case <-dbUpdateChan:
  749. case toScan := <-scanChan:
  750. t.Fatal("Unexpected receive on scanChan:", toScan)
  751. }
  752. info, err = f.mtimefs.Lstat("foo/bar/sym")
  753. if err != nil {
  754. t.Fatal("Unexpected error (file):", err)
  755. }
  756. if info.Owner() != expOwner || info.Group() != expGroup {
  757. t.Fatalf("Expected symlink owner/group to be %d/%d, not %d/%d", expOwner, expGroup, info.Owner(), info.Group())
  758. }
  759. }
  760. // TestSRConflictReplaceFileByDir checks that a conflict is created when an existing file
  761. // is replaced with a directory and versions are conflicting
  762. func TestSRConflictReplaceFileByDir(t *testing.T) {
  763. m, f, wcfgCancel := setupSendReceiveFolder(t)
  764. defer cleanupSRFolder(f, m, wcfgCancel)
  765. ffs := f.Filesystem(nil)
  766. name := "foo"
  767. // create local file
  768. file := createEmptyFileInfo(t, name, ffs)
  769. file.Version = protocol.Vector{}.Update(myID.Short())
  770. f.updateLocalsFromScanning([]protocol.FileInfo{file})
  771. // Simulate remote creating a dir with the same name
  772. file.Type = protocol.FileInfoTypeDirectory
  773. rem := device1.Short()
  774. file.Version = protocol.Vector{}.Update(rem)
  775. file.ModifiedBy = rem
  776. dbUpdateChan := make(chan dbUpdateJob, 1)
  777. scanChan := make(chan string, 1)
  778. f.handleDir(file, fsetSnapshot(t, f.fset), dbUpdateChan, scanChan)
  779. if confls := existingConflicts(name, ffs); len(confls) != 1 {
  780. t.Fatal("Expected one conflict, got", len(confls))
  781. } else if scan := <-scanChan; confls[0] != scan {
  782. t.Fatal("Expected request to scan", confls[0], "got", scan)
  783. }
  784. }
  785. // TestSRConflictReplaceFileByLink checks that a conflict is created when an existing file
  786. // is replaced with a link and versions are conflicting
  787. func TestSRConflictReplaceFileByLink(t *testing.T) {
  788. m, f, wcfgCancel := setupSendReceiveFolder(t)
  789. defer cleanupSRFolder(f, m, wcfgCancel)
  790. ffs := f.Filesystem(nil)
  791. name := "foo"
  792. // create local file
  793. file := createEmptyFileInfo(t, name, ffs)
  794. file.Version = protocol.Vector{}.Update(myID.Short())
  795. f.updateLocalsFromScanning([]protocol.FileInfo{file})
  796. // Simulate remote creating a symlink with the same name
  797. file.Type = protocol.FileInfoTypeSymlink
  798. file.SymlinkTarget = "bar"
  799. rem := device1.Short()
  800. file.Version = protocol.Vector{}.Update(rem)
  801. file.ModifiedBy = rem
  802. dbUpdateChan := make(chan dbUpdateJob, 1)
  803. scanChan := make(chan string, 1)
  804. f.handleSymlink(file, fsetSnapshot(t, f.fset), dbUpdateChan, scanChan)
  805. if confls := existingConflicts(name, ffs); len(confls) != 1 {
  806. t.Fatal("Expected one conflict, got", len(confls))
  807. } else if scan := <-scanChan; confls[0] != scan {
  808. t.Fatal("Expected request to scan", confls[0], "got", scan)
  809. }
  810. }
  811. // TestDeleteBehindSymlink checks that we don't delete or schedule a scan
  812. // when trying to delete a file behind a symlink.
  813. func TestDeleteBehindSymlink(t *testing.T) {
  814. m, f, wcfgCancel := setupSendReceiveFolder(t)
  815. defer cleanupSRFolder(f, m, wcfgCancel)
  816. ffs := f.Filesystem(nil)
  817. destDir := t.TempDir()
  818. destFs := fs.NewFilesystem(fs.FilesystemTypeBasic, destDir)
  819. link := "link"
  820. file := filepath.Join(link, "file")
  821. must(t, ffs.MkdirAll(link, 0755))
  822. fi := createEmptyFileInfo(t, file, ffs)
  823. f.updateLocalsFromScanning([]protocol.FileInfo{fi})
  824. must(t, osutil.RenameOrCopy(fs.CopyRangeMethodStandard, ffs, destFs, file, "file"))
  825. must(t, ffs.RemoveAll(link))
  826. if err := fs.DebugSymlinkForTestsOnly(destFs, ffs, "", link); err != nil {
  827. if build.IsWindows {
  828. // Probably we require permissions we don't have.
  829. t.Skip("Need admin permissions or developer mode to run symlink test on Windows: " + err.Error())
  830. } else {
  831. t.Fatal(err)
  832. }
  833. }
  834. fi.Deleted = true
  835. fi.Version = fi.Version.Update(device1.Short())
  836. scanChan := make(chan string, 1)
  837. dbUpdateChan := make(chan dbUpdateJob, 1)
  838. f.deleteFile(fi, fsetSnapshot(t, f.fset), dbUpdateChan, scanChan)
  839. select {
  840. case f := <-scanChan:
  841. t.Fatalf("Received %v on scanChan", f)
  842. case u := <-dbUpdateChan:
  843. if u.jobType != dbUpdateDeleteFile {
  844. t.Errorf("Expected jobType %v, got %v", dbUpdateDeleteFile, u.jobType)
  845. }
  846. if u.file.Name != fi.Name {
  847. t.Errorf("Expected update for %v, got %v", fi.Name, u.file.Name)
  848. }
  849. default:
  850. t.Fatalf("No db update received")
  851. }
  852. if _, err := destFs.Stat("file"); err != nil {
  853. t.Errorf("Expected no error when stating file behind symlink, got %v", err)
  854. }
  855. }
  856. // Reproduces https://github.com/syncthing/syncthing/issues/6559
  857. func TestPullCtxCancel(t *testing.T) {
  858. m, f, wcfgCancel := setupSendReceiveFolder(t)
  859. defer cleanupSRFolder(f, m, wcfgCancel)
  860. pullChan := make(chan pullBlockState)
  861. finisherChan := make(chan *sharedPullerState)
  862. var cancel context.CancelFunc
  863. f.ctx, cancel = context.WithCancel(context.Background())
  864. go f.pullerRoutine(fsetSnapshot(t, f.fset), pullChan, finisherChan)
  865. defer close(pullChan)
  866. emptyState := func() pullBlockState {
  867. return pullBlockState{
  868. sharedPullerState: newSharedPullerState(protocol.FileInfo{}, nil, f.folderID, "", nil, nil, false, false, protocol.FileInfo{}, false, false),
  869. block: protocol.BlockInfo{},
  870. }
  871. }
  872. cancel()
  873. done := make(chan struct{})
  874. defer close(done)
  875. for i := 0; i < 2; i++ {
  876. go func() {
  877. select {
  878. case pullChan <- emptyState():
  879. case <-done:
  880. }
  881. }()
  882. select {
  883. case s := <-finisherChan:
  884. if s.failed() == nil {
  885. t.Errorf("state %v not failed", i)
  886. }
  887. case <-time.After(5 * time.Second):
  888. t.Fatalf("timed out before receiving state %v on finisherChan", i)
  889. }
  890. }
  891. }
  892. func TestPullDeleteUnscannedDir(t *testing.T) {
  893. m, f, wcfgCancel := setupSendReceiveFolder(t)
  894. defer cleanupSRFolder(f, m, wcfgCancel)
  895. ffs := f.Filesystem(nil)
  896. dir := "foobar"
  897. must(t, ffs.MkdirAll(dir, 0777))
  898. fi := protocol.FileInfo{
  899. Name: dir,
  900. }
  901. scanChan := make(chan string, 1)
  902. dbUpdateChan := make(chan dbUpdateJob, 1)
  903. f.deleteDir(fi, fsetSnapshot(t, f.fset), dbUpdateChan, scanChan)
  904. if _, err := ffs.Stat(dir); fs.IsNotExist(err) {
  905. t.Error("directory has been deleted")
  906. }
  907. select {
  908. case toScan := <-scanChan:
  909. if toScan != dir {
  910. t.Errorf("expected %v to be scanned, got %v", dir, toScan)
  911. }
  912. default:
  913. t.Error("nothing was scheduled for scanning")
  914. }
  915. }
  916. func TestPullCaseOnlyPerformFinish(t *testing.T) {
  917. m, f, wcfgCancel := setupSendReceiveFolder(t)
  918. defer cleanupSRFolder(f, m, wcfgCancel)
  919. ffs := f.Filesystem(nil)
  920. name := "foo"
  921. contents := []byte("contents")
  922. writeFile(t, ffs, name, contents)
  923. must(t, f.scanSubdirs(nil))
  924. var cur protocol.FileInfo
  925. hasCur := false
  926. snap := dbSnapshot(t, m, f.ID)
  927. defer snap.Release()
  928. snap.WithHave(protocol.LocalDeviceID, func(i protocol.FileIntf) bool {
  929. if hasCur {
  930. t.Fatal("got more than one file")
  931. }
  932. cur = i.(protocol.FileInfo)
  933. hasCur = true
  934. return true
  935. })
  936. if !hasCur {
  937. t.Fatal("file is missing")
  938. }
  939. remote := cur
  940. remote.Version = protocol.Vector{}.Update(device1.Short())
  941. remote.Name = strings.ToUpper(cur.Name)
  942. temp := fs.TempName(remote.Name)
  943. writeFile(t, ffs, temp, contents)
  944. scanChan := make(chan string, 1)
  945. dbUpdateChan := make(chan dbUpdateJob, 1)
  946. err := f.performFinish(remote, cur, hasCur, temp, snap, dbUpdateChan, scanChan)
  947. select {
  948. case <-dbUpdateChan: // boring case sensitive filesystem
  949. return
  950. case <-scanChan:
  951. t.Error("no need to scan anything here")
  952. default:
  953. }
  954. var caseErr *fs.ErrCaseConflict
  955. if !errors.As(err, &caseErr) {
  956. t.Error("Expected case conflict error, got", err)
  957. }
  958. }
  959. func TestPullCaseOnlyDir(t *testing.T) {
  960. testPullCaseOnlyDirOrSymlink(t, true)
  961. }
  962. func TestPullCaseOnlySymlink(t *testing.T) {
  963. if build.IsWindows {
  964. t.Skip("symlinks not supported on windows")
  965. }
  966. testPullCaseOnlyDirOrSymlink(t, false)
  967. }
  968. func testPullCaseOnlyDirOrSymlink(t *testing.T, dir bool) {
  969. m, f, wcfgCancel := setupSendReceiveFolder(t)
  970. defer cleanupSRFolder(f, m, wcfgCancel)
  971. ffs := f.Filesystem(nil)
  972. name := "foo"
  973. if dir {
  974. must(t, ffs.Mkdir(name, 0777))
  975. } else {
  976. must(t, ffs.CreateSymlink("target", name))
  977. }
  978. must(t, f.scanSubdirs(nil))
  979. var cur protocol.FileInfo
  980. hasCur := false
  981. snap := dbSnapshot(t, m, f.ID)
  982. defer snap.Release()
  983. snap.WithHave(protocol.LocalDeviceID, func(i protocol.FileIntf) bool {
  984. if hasCur {
  985. t.Fatal("got more than one file")
  986. }
  987. cur = i.(protocol.FileInfo)
  988. hasCur = true
  989. return true
  990. })
  991. if !hasCur {
  992. t.Fatal("file is missing")
  993. }
  994. scanChan := make(chan string, 1)
  995. dbUpdateChan := make(chan dbUpdateJob, 1)
  996. remote := cur
  997. remote.Version = protocol.Vector{}.Update(device1.Short())
  998. remote.Name = strings.ToUpper(cur.Name)
  999. if dir {
  1000. f.handleDir(remote, snap, dbUpdateChan, scanChan)
  1001. } else {
  1002. f.handleSymlink(remote, snap, dbUpdateChan, scanChan)
  1003. }
  1004. select {
  1005. case <-dbUpdateChan: // boring case sensitive filesystem
  1006. return
  1007. case <-scanChan:
  1008. t.Error("no need to scan anything here")
  1009. default:
  1010. }
  1011. if errStr, ok := f.tempPullErrors[remote.Name]; !ok {
  1012. t.Error("missing error for", remote.Name)
  1013. } else if !strings.Contains(errStr, "uses different upper or lowercase") {
  1014. t.Error("unexpected error", errStr, "for", remote.Name)
  1015. }
  1016. }
  1017. func TestPullTempFileCaseConflict(t *testing.T) {
  1018. m, f, wcfgCancel := setupSendReceiveFolder(t)
  1019. defer cleanupSRFolder(f, m, wcfgCancel)
  1020. copyChan := make(chan copyBlocksState, 1)
  1021. file := protocol.FileInfo{Name: "foo"}
  1022. confl := "Foo"
  1023. tempNameConfl := fs.TempName(confl)
  1024. if fd, err := f.mtimefs.Create(tempNameConfl); err != nil {
  1025. t.Fatal(err)
  1026. } else {
  1027. if _, err := fd.Write([]byte("data")); err != nil {
  1028. t.Fatal(err)
  1029. }
  1030. fd.Close()
  1031. }
  1032. f.handleFile(file, fsetSnapshot(t, f.fset), copyChan)
  1033. cs := <-copyChan
  1034. if _, err := cs.tempFile(); err != nil {
  1035. t.Error(err)
  1036. } else {
  1037. cs.finalClose()
  1038. }
  1039. }
  1040. func TestPullCaseOnlyRename(t *testing.T) {
  1041. m, f, wcfgCancel := setupSendReceiveFolder(t)
  1042. defer cleanupSRFolder(f, m, wcfgCancel)
  1043. // tempNameConfl := fs.TempName(confl)
  1044. name := "foo"
  1045. if fd, err := f.mtimefs.Create(name); err != nil {
  1046. t.Fatal(err)
  1047. } else {
  1048. if _, err := fd.Write([]byte("data")); err != nil {
  1049. t.Fatal(err)
  1050. }
  1051. fd.Close()
  1052. }
  1053. must(t, f.scanSubdirs(nil))
  1054. cur, ok := m.testCurrentFolderFile(f.ID, name)
  1055. if !ok {
  1056. t.Fatal("file missing")
  1057. }
  1058. deleted := cur
  1059. deleted.SetDeleted(myID.Short())
  1060. confl := cur
  1061. confl.Name = "Foo"
  1062. confl.Version = confl.Version.Update(device1.Short())
  1063. dbUpdateChan := make(chan dbUpdateJob, 2)
  1064. scanChan := make(chan string, 2)
  1065. snap := fsetSnapshot(t, f.fset)
  1066. defer snap.Release()
  1067. if err := f.renameFile(cur, deleted, confl, snap, dbUpdateChan, scanChan); err != nil {
  1068. t.Error(err)
  1069. }
  1070. }
  1071. func TestPullSymlinkOverExistingWindows(t *testing.T) {
  1072. if !build.IsWindows {
  1073. t.Skip()
  1074. }
  1075. m, f, wcfgCancel := setupSendReceiveFolder(t)
  1076. defer cleanupSRFolder(f, m, wcfgCancel)
  1077. addFakeConn(m, device1, f.ID)
  1078. name := "foo"
  1079. if fd, err := f.mtimefs.Create(name); err != nil {
  1080. t.Fatal(err)
  1081. } else {
  1082. if _, err := fd.Write([]byte("data")); err != nil {
  1083. t.Fatal(err)
  1084. }
  1085. fd.Close()
  1086. }
  1087. must(t, f.scanSubdirs(nil))
  1088. file, ok := m.testCurrentFolderFile(f.ID, name)
  1089. if !ok {
  1090. t.Fatal("file missing")
  1091. }
  1092. must(t, m.Index(device1, f.ID, []protocol.FileInfo{{Name: name, Type: protocol.FileInfoTypeSymlink, Version: file.Version.Update(device1.Short())}}))
  1093. scanChan := make(chan string)
  1094. changed, err := f.pullerIteration(scanChan)
  1095. must(t, err)
  1096. if changed != 1 {
  1097. t.Error("Expected one change in pull, got", changed)
  1098. }
  1099. if file, ok := m.testCurrentFolderFile(f.ID, name); !ok {
  1100. t.Error("symlink entry missing")
  1101. } else if !file.IsUnsupported() {
  1102. t.Error("symlink entry isn't marked as unsupported")
  1103. }
  1104. if _, err := f.mtimefs.Lstat(name); err == nil {
  1105. t.Error("old file still exists on disk")
  1106. } else if !fs.IsNotExist(err) {
  1107. t.Error(err)
  1108. }
  1109. }
  1110. func TestPullDeleteCaseConflict(t *testing.T) {
  1111. m, f, wcfgCancel := setupSendReceiveFolder(t)
  1112. defer cleanupSRFolder(f, m, wcfgCancel)
  1113. name := "foo"
  1114. fi := protocol.FileInfo{Name: "Foo"}
  1115. dbUpdateChan := make(chan dbUpdateJob, 1)
  1116. scanChan := make(chan string)
  1117. if fd, err := f.mtimefs.Create(name); err != nil {
  1118. t.Fatal(err)
  1119. } else {
  1120. if _, err := fd.Write([]byte("data")); err != nil {
  1121. t.Fatal(err)
  1122. }
  1123. fd.Close()
  1124. }
  1125. f.deleteFileWithCurrent(fi, protocol.FileInfo{}, false, dbUpdateChan, scanChan)
  1126. select {
  1127. case <-dbUpdateChan:
  1128. default:
  1129. t.Error("Missing db update for file")
  1130. }
  1131. snap := fsetSnapshot(t, f.fset)
  1132. defer snap.Release()
  1133. f.deleteDir(fi, snap, dbUpdateChan, scanChan)
  1134. select {
  1135. case <-dbUpdateChan:
  1136. default:
  1137. t.Error("Missing db update for dir")
  1138. }
  1139. }
  1140. func TestPullDeleteIgnoreChildDir(t *testing.T) {
  1141. m, f, wcfgCancel := setupSendReceiveFolder(t)
  1142. defer cleanupSRFolder(f, m, wcfgCancel)
  1143. parent := "parent"
  1144. del := "ignored"
  1145. child := "keep"
  1146. matcher := ignore.New(f.mtimefs)
  1147. must(t, matcher.Parse(bytes.NewBufferString(fmt.Sprintf(`
  1148. !%v
  1149. (?d)%v
  1150. `, child, del)), ""))
  1151. f.ignores = matcher
  1152. must(t, f.mtimefs.Mkdir(parent, 0777))
  1153. must(t, f.mtimefs.Mkdir(filepath.Join(parent, del), 0777))
  1154. must(t, f.mtimefs.Mkdir(filepath.Join(parent, del, child), 0777))
  1155. scanChan := make(chan string, 2)
  1156. err := f.deleteDirOnDisk(parent, fsetSnapshot(t, f.fset), scanChan)
  1157. if err == nil {
  1158. t.Error("no error")
  1159. }
  1160. }
  1161. func cleanupSharedPullerState(s *sharedPullerState) {
  1162. s.mut.Lock()
  1163. defer s.mut.Unlock()
  1164. if s.writer == nil {
  1165. return
  1166. }
  1167. s.writer.mut.Lock()
  1168. s.writer.fd.Close()
  1169. s.writer.mut.Unlock()
  1170. }
  1171. func startCopier(f *sendReceiveFolder, pullChan chan<- pullBlockState, finisherChan chan<- *sharedPullerState) (chan copyBlocksState, sync.WaitGroup) {
  1172. copyChan := make(chan copyBlocksState)
  1173. wg := sync.NewWaitGroup()
  1174. wg.Add(1)
  1175. go func() {
  1176. f.copierRoutine(copyChan, pullChan, finisherChan)
  1177. wg.Done()
  1178. }()
  1179. return copyChan, wg
  1180. }