folder_sendrecv_test.go 36 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255
  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. "errors"
  11. "fmt"
  12. "io"
  13. "os"
  14. "path/filepath"
  15. "runtime/pprof"
  16. "strconv"
  17. "strings"
  18. "sync"
  19. "testing"
  20. "time"
  21. "github.com/syncthing/syncthing/internal/itererr"
  22. "github.com/syncthing/syncthing/lib/build"
  23. "github.com/syncthing/syncthing/lib/config"
  24. "github.com/syncthing/syncthing/lib/events"
  25. "github.com/syncthing/syncthing/lib/fs"
  26. "github.com/syncthing/syncthing/lib/ignore"
  27. "github.com/syncthing/syncthing/lib/protocol"
  28. "github.com/syncthing/syncthing/lib/scanner"
  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. func prepareTmpFile(to fs.Filesystem) (string, error) {
  42. tmpName := fs.TempName("file")
  43. in, err := os.Open("testdata/tmpfile")
  44. if err != nil {
  45. return "", err
  46. }
  47. defer in.Close()
  48. out, err := to.Create(tmpName)
  49. if err != nil {
  50. return "", err
  51. }
  52. defer out.Close()
  53. if _, err = io.Copy(out, in); err != nil {
  54. return "", err
  55. }
  56. future := time.Now().Add(time.Hour)
  57. if err := to.Chtimes(tmpName, future, future); err != nil {
  58. return "", err
  59. }
  60. return tmpName, nil
  61. }
  62. var diffTestData = []struct {
  63. a string
  64. b string
  65. s int
  66. d []protocol.BlockInfo
  67. }{
  68. {"contents", "contents", 1024, []protocol.BlockInfo{}},
  69. {"", "", 1024, []protocol.BlockInfo{}},
  70. {"contents", "contents", 3, []protocol.BlockInfo{}},
  71. {"contents", "cantents", 3, []protocol.BlockInfo{{Offset: 0, Size: 3}}},
  72. {"contents", "contants", 3, []protocol.BlockInfo{{Offset: 3, Size: 3}}},
  73. {"contents", "cantants", 3, []protocol.BlockInfo{{Offset: 0, Size: 3}, {Offset: 3, Size: 3}}},
  74. {"contents", "", 3, []protocol.BlockInfo{{Offset: 0, Size: 0}}},
  75. {"", "contents", 3, []protocol.BlockInfo{{Offset: 0, Size: 3}, {Offset: 3, Size: 3}, {Offset: 6, Size: 2}}},
  76. {"con", "contents", 3, []protocol.BlockInfo{{Offset: 3, Size: 3}, {Offset: 6, Size: 2}}},
  77. {"contents", "con", 3, nil},
  78. {"contents", "cont", 3, []protocol.BlockInfo{{Offset: 3, Size: 1}}},
  79. {"cont", "contents", 3, []protocol.BlockInfo{{Offset: 3, Size: 3}, {Offset: 6, Size: 2}}},
  80. }
  81. func setupFile(filename string, blockNumbers []int) protocol.FileInfo {
  82. // Create existing file
  83. existingBlocks := make([]protocol.BlockInfo, len(blockNumbers))
  84. for i := range blockNumbers {
  85. existingBlocks[i] = blocks[blockNumbers[i]]
  86. }
  87. return protocol.FileInfo{
  88. Name: filename,
  89. Blocks: existingBlocks,
  90. }
  91. }
  92. func createEmptyFileInfo(t *testing.T, name string, fs fs.Filesystem) protocol.FileInfo {
  93. t.Helper()
  94. writeFile(t, fs, name, nil)
  95. fi, err := fs.Stat(name)
  96. must(t, err)
  97. file, err := scanner.CreateFileInfo(fi, name, fs, false, false, config.XattrFilter{})
  98. must(t, err)
  99. return file
  100. }
  101. // Sets up a folder and model, but makes sure the services aren't actually running.
  102. func setupSendReceiveFolder(t testing.TB, files ...protocol.FileInfo) (*testModel, *sendReceiveFolder) {
  103. w, fcfg := newDefaultCfgWrapper(t)
  104. // Initialise model and stop immediately.
  105. model := setupModel(t, w)
  106. model.cancel()
  107. <-model.stopped
  108. r, _ := model.folderRunners.Get(fcfg.ID)
  109. f := r.(*sendReceiveFolder)
  110. f.tempPullErrors = make(map[string]string)
  111. // Update index
  112. if files != nil {
  113. f.updateLocalsFromScanning(files)
  114. }
  115. return model, f
  116. }
  117. // Layout of the files: (indexes from the above array)
  118. // 12345678 - Required file
  119. // 02005008 - Existing file (currently in the index)
  120. // 02340070 - Temp file on the disk
  121. func TestHandleFile(t *testing.T) {
  122. // After the diff between required and existing we should:
  123. // Copy: 2, 5, 8
  124. // Pull: 1, 3, 4, 6, 7
  125. existingBlocks := []int{0, 2, 0, 0, 5, 0, 0, 8}
  126. existingFile := setupFile("filex", existingBlocks)
  127. requiredFile := existingFile
  128. requiredFile.Blocks = blocks[1:]
  129. _, f := setupSendReceiveFolder(t, existingFile)
  130. copyChan := make(chan copyBlocksState, 1)
  131. f.handleFile(t.Context(), requiredFile, copyChan)
  132. // Receive the results
  133. toCopy := <-copyChan
  134. if len(toCopy.blocks) != 8 {
  135. t.Errorf("Unexpected count of copy blocks: %d != 8", len(toCopy.blocks))
  136. }
  137. for _, block := range blocks[1:] {
  138. found := false
  139. for _, toCopyBlock := range toCopy.blocks {
  140. if bytes.Equal(toCopyBlock.Hash, block.Hash) {
  141. found = true
  142. break
  143. }
  144. }
  145. if !found {
  146. t.Errorf("Did not find block %s", block.String())
  147. }
  148. }
  149. }
  150. func TestHandleFileWithTemp(t *testing.T) {
  151. // After diff between required and existing we should:
  152. // Copy: 2, 5, 8
  153. // Pull: 1, 3, 4, 6, 7
  154. // After dropping out blocks already on the temp file we should:
  155. // Copy: 5, 8
  156. // Pull: 1, 6
  157. existingBlocks := []int{0, 2, 0, 0, 5, 0, 0, 8}
  158. existingFile := setupFile("file", existingBlocks)
  159. requiredFile := existingFile
  160. requiredFile.Blocks = blocks[1:]
  161. _, f := setupSendReceiveFolder(t, existingFile)
  162. if _, err := prepareTmpFile(f.Filesystem()); err != nil {
  163. t.Fatal(err)
  164. }
  165. copyChan := make(chan copyBlocksState, 1)
  166. f.handleFile(t.Context(), requiredFile, copyChan)
  167. // Receive the results
  168. toCopy := <-copyChan
  169. if len(toCopy.blocks) != 4 {
  170. t.Errorf("Unexpected count of copy blocks: %d != 4", len(toCopy.blocks))
  171. }
  172. for _, idx := range []int{1, 5, 6, 8} {
  173. found := false
  174. block := blocks[idx]
  175. for _, toCopyBlock := range toCopy.blocks {
  176. if bytes.Equal(toCopyBlock.Hash, block.Hash) {
  177. found = true
  178. break
  179. }
  180. }
  181. if !found {
  182. t.Errorf("Did not find block %s", block.String())
  183. }
  184. }
  185. }
  186. func TestCopierFinder(t *testing.T) {
  187. // After diff between required and existing we should:
  188. // Copy: 1, 2, 3, 4, 6, 7, 8
  189. // Since there is no existing file, nor a temp file
  190. // After dropping out blocks found locally:
  191. // Pull: 1, 5, 6, 8
  192. tempFile := fs.TempName("file2")
  193. existingBlocks := []int{0, 2, 3, 4, 0, 0, 7, 0}
  194. existingFile := setupFile(fs.TempName("file"), existingBlocks)
  195. existingFile.Size = 1
  196. requiredFile := existingFile
  197. requiredFile.Blocks = blocks[1:]
  198. requiredFile.Name = "file2"
  199. _, f := setupSendReceiveFolder(t, existingFile)
  200. if _, err := prepareTmpFile(f.Filesystem()); err != nil {
  201. t.Fatal(err)
  202. }
  203. copyChan := make(chan copyBlocksState)
  204. pullChan := make(chan pullBlockState, 4)
  205. finisherChan := make(chan *sharedPullerState, 1)
  206. // Run a single fetcher routine
  207. go f.copierRoutine(t.Context(), copyChan, pullChan, finisherChan)
  208. defer close(copyChan)
  209. f.handleFile(t.Context(), requiredFile, copyChan)
  210. timeout := time.After(10 * time.Second)
  211. pulls := make([]pullBlockState, 4)
  212. for i := 0; i < 4; i++ {
  213. select {
  214. case pulls[i] = <-pullChan:
  215. case <-timeout:
  216. t.Fatalf("Timed out before receiving all 4 states on pullChan (already got %v)", i)
  217. }
  218. }
  219. var finish *sharedPullerState
  220. select {
  221. case finish = <-finisherChan:
  222. case <-timeout:
  223. t.Fatal("Timed out before receiving 4 states on pullChan")
  224. }
  225. defer cleanupSharedPullerState(finish)
  226. select {
  227. case v := <-pullChan:
  228. t.Logf("%+v\n", v)
  229. t.Fatal("Pull channel had data to be read")
  230. case <-finisherChan:
  231. t.Fatal("Finisher channel has data to be read")
  232. default:
  233. }
  234. // Verify that the right blocks went into the pull list.
  235. // They are pulled in random order.
  236. for _, idx := range []int{1, 5, 6, 8} {
  237. found := false
  238. block := blocks[idx]
  239. for _, pulledBlock := range pulls {
  240. if bytes.Equal(pulledBlock.block.Hash, block.Hash) {
  241. found = true
  242. break
  243. }
  244. }
  245. if !found {
  246. t.Errorf("Did not find block %s", block.String())
  247. }
  248. if !bytes.Equal(finish.file.Blocks[idx-1].Hash, blocks[idx].Hash) {
  249. t.Errorf("Block %d mismatch: %s != %s", idx, finish.file.Blocks[idx-1].String(), blocks[idx].String())
  250. }
  251. }
  252. // Verify that the fetched blocks have actually been written to the temp file
  253. blks, err := scanner.HashFile(t.Context(), f.ID, f.Filesystem(), tempFile, protocol.MinBlockSize, nil)
  254. if err != nil {
  255. t.Log(err)
  256. }
  257. for _, eq := range []int{2, 3, 4, 7} {
  258. if !bytes.Equal(blks[eq-1].Hash, blocks[eq].Hash) {
  259. t.Errorf("Block %d mismatch: %s != %s", eq, blks[eq-1].String(), blocks[eq].String())
  260. }
  261. }
  262. }
  263. // Test that updating a file removes its old blocks from the blockmap
  264. func TestCopierCleanup(t *testing.T) {
  265. // Create a file
  266. file := setupFile("test", []int{0})
  267. file.Size = 1
  268. m, f := setupSendReceiveFolder(t, file)
  269. file.Blocks = []protocol.BlockInfo{blocks[1]}
  270. file.Version = file.Version.Update(myID.Short())
  271. // Update index (removing old blocks)
  272. f.updateLocalsFromScanning([]protocol.FileInfo{file})
  273. if vals, err := itererr.Collect(m.sdb.AllLocalBlocksWithHash(f.ID, blocks[0].Hash)); err != nil || len(vals) > 0 {
  274. t.Error("Unexpected block found")
  275. }
  276. if vals, err := itererr.Collect(m.sdb.AllLocalBlocksWithHash(f.ID, blocks[1].Hash)); err != nil || len(vals) == 0 {
  277. t.Error("Expected block not found")
  278. }
  279. file.Blocks = []protocol.BlockInfo{blocks[0]}
  280. file.Version = file.Version.Update(myID.Short())
  281. // Update index (removing old blocks)
  282. f.updateLocalsFromScanning([]protocol.FileInfo{file})
  283. if vals, err := itererr.Collect(m.sdb.AllLocalBlocksWithHash(f.ID, blocks[0].Hash)); err != nil || len(vals) == 0 {
  284. t.Error("Unexpected block found")
  285. }
  286. if vals, err := itererr.Collect(m.sdb.AllLocalBlocksWithHash(f.ID, blocks[1].Hash)); err != nil || len(vals) > 0 {
  287. t.Error("Expected block not found")
  288. }
  289. }
  290. func TestDeregisterOnFailInCopy(t *testing.T) {
  291. file := setupFile("filex", []int{0, 2, 0, 0, 5, 0, 0, 8})
  292. m, f := setupSendReceiveFolder(t)
  293. // Set up our evet subscription early
  294. s := m.evLogger.Subscribe(events.ItemFinished)
  295. // queue.Done should be called by the finisher routine
  296. f.queue.Push("filex", 0, time.Time{})
  297. f.queue.Pop()
  298. if f.queue.lenProgress() != 1 {
  299. t.Fatal("Expected file in progress")
  300. }
  301. pullChan := make(chan pullBlockState)
  302. finisherBufferChan := make(chan *sharedPullerState, 1)
  303. finisherChan := make(chan *sharedPullerState)
  304. dbUpdateChan := make(chan dbUpdateJob, 1)
  305. copyChan, copyWg := startCopier(t.Context(), f, pullChan, finisherBufferChan)
  306. go f.finisherRoutine(t.Context(), finisherChan, dbUpdateChan, make(chan string))
  307. defer func() {
  308. close(copyChan)
  309. copyWg.Wait()
  310. close(pullChan)
  311. close(finisherBufferChan)
  312. close(finisherChan)
  313. }()
  314. f.handleFile(t.Context(), file, copyChan)
  315. // Receive a block at puller, to indicate that at least a single copier
  316. // loop has been performed.
  317. var toPull pullBlockState
  318. select {
  319. case toPull = <-pullChan:
  320. case <-time.After(10 * time.Second):
  321. t.Fatal("timed out")
  322. }
  323. // Unblock copier
  324. go func() {
  325. for range pullChan {
  326. }
  327. }()
  328. // Close the file, causing errors on further access
  329. toPull.sharedPullerState.fail(os.ErrNotExist)
  330. select {
  331. case state := <-finisherBufferChan:
  332. // At this point the file should still be registered with both the job
  333. // queue, and the progress emitter. Verify this.
  334. if f.model.progressEmitter.lenRegistry() != 1 || f.queue.lenProgress() != 1 || f.queue.lenQueued() != 0 {
  335. t.Fatal("Could not find file")
  336. }
  337. // Pass the file down the real finisher, and give it time to consume
  338. finisherChan <- state
  339. t0 := time.Now()
  340. if ev, err := s.Poll(time.Minute); err != nil {
  341. t.Fatal("Got error waiting for ItemFinished event:", err)
  342. } else if n := ev.Data.(map[string]interface{})["item"]; n != state.file.Name {
  343. t.Fatal("Got ItemFinished event for wrong file:", n)
  344. }
  345. t.Log("event took", time.Since(t0))
  346. state.mut.Lock()
  347. stateWriter := state.writer
  348. state.mut.Unlock()
  349. if stateWriter != nil {
  350. t.Fatal("File not closed?")
  351. }
  352. if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
  353. t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
  354. }
  355. // Doing it again should have no effect
  356. finisherChan <- state
  357. if _, err := s.Poll(time.Second); err != events.ErrTimeout {
  358. t.Fatal("Expected timeout, not another event", err)
  359. }
  360. if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
  361. t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
  362. }
  363. case <-time.After(5 * time.Second):
  364. t.Fatal("Didn't get anything to the finisher")
  365. }
  366. }
  367. func TestDeregisterOnFailInPull(t *testing.T) {
  368. file := setupFile("filex", []int{0, 2, 0, 0, 5, 0, 0, 8})
  369. m, f := setupSendReceiveFolder(t)
  370. // Set up our evet subscription early
  371. s := m.evLogger.Subscribe(events.ItemFinished)
  372. // queue.Done should be called by the finisher routine
  373. f.queue.Push("filex", 0, time.Time{})
  374. f.queue.Pop()
  375. if f.queue.lenProgress() != 1 {
  376. t.Fatal("Expected file in progress")
  377. }
  378. pullChan := make(chan pullBlockState)
  379. finisherBufferChan := make(chan *sharedPullerState)
  380. finisherChan := make(chan *sharedPullerState)
  381. dbUpdateChan := make(chan dbUpdateJob, 1)
  382. copyChan, copyWg := startCopier(t.Context(), f, pullChan, finisherBufferChan)
  383. var pullWg sync.WaitGroup
  384. pullWg.Add(1)
  385. go func() {
  386. f.pullerRoutine(t.Context(), pullChan, finisherBufferChan)
  387. pullWg.Done()
  388. }()
  389. go f.finisherRoutine(t.Context(), finisherChan, dbUpdateChan, make(chan string))
  390. defer func() {
  391. // Unblock copier and puller
  392. go func() {
  393. for range finisherBufferChan {
  394. }
  395. }()
  396. close(copyChan)
  397. copyWg.Wait()
  398. close(pullChan)
  399. pullWg.Wait()
  400. close(finisherBufferChan)
  401. close(finisherChan)
  402. }()
  403. f.handleFile(t.Context(), file, copyChan)
  404. // Receive at finisher, we should error out as puller has nowhere to pull
  405. // from.
  406. timeout = time.Second
  407. // Both the puller and copier may send to the finisherBufferChan.
  408. var state *sharedPullerState
  409. after := time.After(5 * time.Second)
  410. for {
  411. select {
  412. case state = <-finisherBufferChan:
  413. case <-after:
  414. t.Fatal("Didn't get failed state to the finisher")
  415. }
  416. if state.failed() != nil {
  417. break
  418. }
  419. }
  420. // At this point the file should still be registered with both the job
  421. // queue, and the progress emitter. Verify this.
  422. if f.model.progressEmitter.lenRegistry() != 1 || f.queue.lenProgress() != 1 || f.queue.lenQueued() != 0 {
  423. t.Fatal("Could not find file")
  424. }
  425. // Pass the file down the real finisher, and give it time to consume
  426. finisherChan <- state
  427. t0 := time.Now()
  428. if ev, err := s.Poll(time.Minute); err != nil {
  429. t.Fatal("Got error waiting for ItemFinished event:", err)
  430. } else if n := ev.Data.(map[string]interface{})["item"]; n != state.file.Name {
  431. t.Fatal("Got ItemFinished event for wrong file:", n)
  432. }
  433. t.Log("event took", time.Since(t0))
  434. state.mut.Lock()
  435. stateWriter := state.writer
  436. state.mut.Unlock()
  437. if stateWriter != nil {
  438. t.Fatal("File not closed?")
  439. }
  440. if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
  441. t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
  442. }
  443. // Doing it again should have no effect
  444. finisherChan <- state
  445. if _, err := s.Poll(time.Second); err != events.ErrTimeout {
  446. t.Fatal("Expected timeout, not another event", err)
  447. }
  448. if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
  449. t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
  450. }
  451. }
  452. func TestIssue3164(t *testing.T) {
  453. _, f := setupSendReceiveFolder(t)
  454. ffs := f.Filesystem()
  455. ignDir := filepath.Join("issue3164", "oktodelete")
  456. subDir := filepath.Join(ignDir, "foobar")
  457. must(t, ffs.MkdirAll(subDir, 0o777))
  458. must(t, fs.WriteFile(ffs, filepath.Join(subDir, "file"), []byte("Hello"), 0o644))
  459. must(t, fs.WriteFile(ffs, filepath.Join(ignDir, "file"), []byte("Hello"), 0o644))
  460. file := protocol.FileInfo{
  461. Name: "issue3164",
  462. }
  463. must(t, f.scanSubdirs(t.Context(), nil))
  464. matcher := ignore.New(ffs)
  465. must(t, matcher.Parse(bytes.NewBufferString("(?d)oktodelete"), ""))
  466. f.ignores = matcher
  467. dbUpdateChan := make(chan dbUpdateJob, 1)
  468. f.deleteDir(file, dbUpdateChan, make(chan string))
  469. if _, err := ffs.Stat("issue3164"); !fs.IsNotExist(err) {
  470. t.Fatal(err)
  471. }
  472. }
  473. func TestDiff(t *testing.T) {
  474. for i, test := range diffTestData {
  475. a, _ := scanner.Blocks(t.Context(), bytes.NewBufferString(test.a), test.s, -1, nil)
  476. b, _ := scanner.Blocks(t.Context(), bytes.NewBufferString(test.b), test.s, -1, nil)
  477. _, d := blockDiff(a, b)
  478. if len(d) != len(test.d) {
  479. t.Fatalf("Incorrect length for diff %d; %d != %d", i, len(d), len(test.d))
  480. } else {
  481. for j := range test.d {
  482. if d[j].Offset != test.d[j].Offset {
  483. t.Errorf("Incorrect offset for diff %d block %d; %d != %d", i, j, d[j].Offset, test.d[j].Offset)
  484. }
  485. if d[j].Size != test.d[j].Size {
  486. t.Errorf("Incorrect length for diff %d block %d; %d != %d", i, j, d[j].Size, test.d[j].Size)
  487. }
  488. }
  489. }
  490. }
  491. }
  492. func BenchmarkDiff(b *testing.B) {
  493. testCases := make([]struct{ a, b []protocol.BlockInfo }, 0, len(diffTestData))
  494. for _, test := range diffTestData {
  495. aBlocks, _ := scanner.Blocks(b.Context(), bytes.NewBufferString(test.a), test.s, -1, nil)
  496. bBlocks, _ := scanner.Blocks(b.Context(), bytes.NewBufferString(test.b), test.s, -1, nil)
  497. testCases = append(testCases, struct{ a, b []protocol.BlockInfo }{aBlocks, bBlocks})
  498. }
  499. b.ReportAllocs()
  500. b.ResetTimer()
  501. for i := 0; i < b.N; i++ {
  502. for _, tc := range testCases {
  503. blockDiff(tc.a, tc.b)
  504. }
  505. }
  506. }
  507. func TestDiffEmpty(t *testing.T) {
  508. emptyCases := []struct {
  509. a []protocol.BlockInfo
  510. b []protocol.BlockInfo
  511. need int
  512. have int
  513. }{
  514. {nil, nil, 0, 0},
  515. {[]protocol.BlockInfo{{Offset: 3, Size: 1}}, nil, 0, 0},
  516. {nil, []protocol.BlockInfo{{Offset: 3, Size: 1}}, 1, 0},
  517. }
  518. for _, emptyCase := range emptyCases {
  519. h, n := blockDiff(emptyCase.a, emptyCase.b)
  520. if len(h) != emptyCase.have {
  521. t.Errorf("incorrect have: %d != %d", len(h), emptyCase.have)
  522. }
  523. if len(n) != emptyCase.need {
  524. t.Errorf("incorrect have: %d != %d", len(h), emptyCase.have)
  525. }
  526. }
  527. }
  528. // TestDeleteIgnorePerms checks, that a file gets deleted when the IgnorePerms
  529. // option is true and the permissions do not match between the file on disk and
  530. // in the db.
  531. func TestDeleteIgnorePerms(t *testing.T) {
  532. _, f := setupSendReceiveFolder(t)
  533. ffs := f.Filesystem()
  534. f.IgnorePerms = true
  535. name := "deleteIgnorePerms"
  536. file, err := ffs.Create(name)
  537. if err != nil {
  538. t.Error(err)
  539. }
  540. defer file.Close()
  541. stat, err := file.Stat()
  542. must(t, err)
  543. fi, err := scanner.CreateFileInfo(stat, name, ffs, false, false, config.XattrFilter{})
  544. must(t, err)
  545. ffs.Chmod(name, 0o600)
  546. if info, err := ffs.Stat(name); err == nil {
  547. fi.InodeChangeNs = info.InodeChangeTime().UnixNano()
  548. }
  549. scanChan := make(chan string, 1)
  550. err = f.checkToBeDeleted(fi, fi, true, scanChan)
  551. must(t, err)
  552. }
  553. func TestCopyOwner(t *testing.T) {
  554. // Verifies that owner and group are copied from the parent, for both
  555. // files and directories.
  556. if build.IsWindows {
  557. t.Skip("copying owner not supported on Windows")
  558. }
  559. const (
  560. expOwner = 1234
  561. expGroup = 5678
  562. )
  563. // This test hung on a regression, taking a long time to fail - speed that up.
  564. ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
  565. defer cancel()
  566. go func() {
  567. <-ctx.Done()
  568. if errors.Is(ctx.Err(), context.DeadlineExceeded) {
  569. pprof.Lookup("goroutine").WriteTo(os.Stdout, 2)
  570. panic("timed out before test finished")
  571. }
  572. }()
  573. // Set up a folder with the CopyParentOwner bit and backed by a fake
  574. // filesystem.
  575. m, f := setupSendReceiveFolder(t)
  576. f.folder.FolderConfiguration = newFolderConfiguration(m.cfg, f.ID, f.Label, config.FilesystemTypeFake, "/TestCopyOwner")
  577. f.folder.FolderConfiguration.CopyOwnershipFromParent = true
  578. // Create a parent dir with a certain owner/group.
  579. f.mtimefs.Mkdir("foo", 0o755)
  580. f.mtimefs.Lchown("foo", strconv.Itoa(expOwner), strconv.Itoa(expGroup))
  581. dir := protocol.FileInfo{
  582. Name: "foo/bar",
  583. Type: protocol.FileInfoTypeDirectory,
  584. Permissions: 0o755,
  585. }
  586. // Have the folder create a subdirectory, verify that it's the correct
  587. // owner/group.
  588. dbUpdateChan := make(chan dbUpdateJob, 1)
  589. scanChan := make(chan string)
  590. defer close(dbUpdateChan)
  591. f.handleDir(dir, dbUpdateChan, scanChan)
  592. select {
  593. case <-dbUpdateChan: // empty the channel for later
  594. case toScan := <-scanChan:
  595. t.Fatal("Unexpected receive on scanChan:", toScan)
  596. }
  597. info, err := f.mtimefs.Lstat("foo/bar")
  598. if err != nil {
  599. t.Fatal("Unexpected error (dir):", err)
  600. }
  601. if info.Owner() != expOwner || info.Group() != expGroup {
  602. t.Fatalf("Expected dir owner/group to be %d/%d, not %d/%d", expOwner, expGroup, info.Owner(), info.Group())
  603. }
  604. // Have the folder create a file, verify it's the correct owner/group.
  605. // File is zero sized to avoid having to handle copies/pulls.
  606. file := protocol.FileInfo{
  607. Name: "foo/bar/baz",
  608. Type: protocol.FileInfoTypeFile,
  609. Permissions: 0o644,
  610. }
  611. // Wire some stuff. The flow here is handleFile() -[copierChan]->
  612. // copierRoutine() -[finisherChan]-> finisherRoutine() -[dbUpdateChan]->
  613. // back to us and we're done. The copier routine doesn't do anything,
  614. // but it's the way data is passed around. When the database update
  615. // comes the finisher is done.
  616. finisherChan := make(chan *sharedPullerState)
  617. copierChan, copyWg := startCopier(t.Context(), f, nil, finisherChan)
  618. go f.finisherRoutine(t.Context(), finisherChan, dbUpdateChan, nil)
  619. defer func() {
  620. close(copierChan)
  621. copyWg.Wait()
  622. close(finisherChan)
  623. }()
  624. f.handleFile(t.Context(), file, copierChan)
  625. <-dbUpdateChan
  626. info, err = f.mtimefs.Lstat("foo/bar/baz")
  627. if err != nil {
  628. t.Fatal("Unexpected error (file):", err)
  629. }
  630. if info.Owner() != expOwner || info.Group() != expGroup {
  631. t.Fatalf("Expected file owner/group to be %d/%d, not %d/%d", expOwner, expGroup, info.Owner(), info.Group())
  632. }
  633. // Have the folder create a symlink. Verify it accordingly.
  634. symlink := protocol.FileInfo{
  635. Name: "foo/bar/sym",
  636. Type: protocol.FileInfoTypeSymlink,
  637. Permissions: 0o644,
  638. SymlinkTarget: []byte("over the rainbow"),
  639. }
  640. f.handleSymlink(symlink, dbUpdateChan, scanChan)
  641. select {
  642. case <-dbUpdateChan:
  643. case toScan := <-scanChan:
  644. t.Fatal("Unexpected receive on scanChan:", toScan)
  645. }
  646. info, err = f.mtimefs.Lstat("foo/bar/sym")
  647. if err != nil {
  648. t.Fatal("Unexpected error (file):", err)
  649. }
  650. if info.Owner() != expOwner || info.Group() != expGroup {
  651. t.Fatalf("Expected symlink owner/group to be %d/%d, not %d/%d", expOwner, expGroup, info.Owner(), info.Group())
  652. }
  653. }
  654. // TestSRConflictReplaceFileByDir checks that a conflict is created when an existing file
  655. // is replaced with a directory and versions are conflicting
  656. func TestSRConflictReplaceFileByDir(t *testing.T) {
  657. _, f := setupSendReceiveFolder(t)
  658. ffs := f.Filesystem()
  659. name := "foo"
  660. // create local file
  661. file := createEmptyFileInfo(t, name, ffs)
  662. file.Version = protocol.Vector{}.Update(myID.Short())
  663. f.updateLocalsFromScanning([]protocol.FileInfo{file})
  664. // Simulate remote creating a dir with the same name
  665. file.Type = protocol.FileInfoTypeDirectory
  666. rem := device1.Short()
  667. file.Version = protocol.Vector{}.Update(rem)
  668. file.ModifiedBy = rem
  669. dbUpdateChan := make(chan dbUpdateJob, 1)
  670. scanChan := make(chan string, 1)
  671. f.handleDir(file, dbUpdateChan, scanChan)
  672. if confls := existingConflicts(name, ffs); len(confls) != 1 {
  673. t.Fatal("Expected one conflict, got", len(confls))
  674. } else if scan := <-scanChan; confls[0] != scan {
  675. t.Fatal("Expected request to scan", confls[0], "got", scan)
  676. }
  677. }
  678. // TestSRConflictReplaceFileByLink checks that a conflict is created when an existing file
  679. // is replaced with a link and versions are conflicting
  680. func TestSRConflictReplaceFileByLink(t *testing.T) {
  681. _, f := setupSendReceiveFolder(t)
  682. ffs := f.Filesystem()
  683. name := "foo"
  684. // create local file
  685. file := createEmptyFileInfo(t, name, ffs)
  686. file.Version = protocol.Vector{}.Update(myID.Short())
  687. f.updateLocalsFromScanning([]protocol.FileInfo{file})
  688. // Simulate remote creating a symlink with the same name
  689. file.Type = protocol.FileInfoTypeSymlink
  690. file.SymlinkTarget = []byte("bar")
  691. rem := device1.Short()
  692. file.Version = protocol.Vector{}.Update(rem)
  693. file.ModifiedBy = rem
  694. dbUpdateChan := make(chan dbUpdateJob, 1)
  695. scanChan := make(chan string, 1)
  696. f.handleSymlink(file, dbUpdateChan, scanChan)
  697. if confls := existingConflicts(name, ffs); len(confls) != 1 {
  698. t.Fatal("Expected one conflict, got", len(confls))
  699. } else if scan := <-scanChan; confls[0] != scan {
  700. t.Fatal("Expected request to scan", confls[0], "got", scan)
  701. }
  702. }
  703. // TestDeleteBehindSymlink checks that we don't delete or schedule a scan
  704. // when trying to delete a file behind a symlink.
  705. func TestDeleteBehindSymlink(t *testing.T) {
  706. _, f := setupSendReceiveFolder(t)
  707. ffs := f.Filesystem()
  708. link := "link"
  709. linkFile := filepath.Join(link, "file")
  710. must(t, ffs.MkdirAll(link, 0o755))
  711. fi := createEmptyFileInfo(t, linkFile, ffs)
  712. f.updateLocalsFromScanning([]protocol.FileInfo{fi})
  713. must(t, ffs.Rename(linkFile, "file"))
  714. must(t, ffs.RemoveAll(link))
  715. must(t, ffs.CreateSymlink("/", link))
  716. fi.Deleted = true
  717. fi.Version = fi.Version.Update(device1.Short())
  718. scanChan := make(chan string, 1)
  719. dbUpdateChan := make(chan dbUpdateJob, 1)
  720. f.deleteFile(fi, dbUpdateChan, scanChan)
  721. select {
  722. case f := <-scanChan:
  723. t.Fatalf("Received %v on scanChan", f)
  724. case u := <-dbUpdateChan:
  725. if u.jobType != dbUpdateDeleteFile {
  726. t.Errorf("Expected jobType %v, got %v", dbUpdateDeleteFile, u.jobType)
  727. }
  728. if u.file.Name != fi.Name {
  729. t.Errorf("Expected update for %v, got %v", fi.Name, u.file.Name)
  730. }
  731. default:
  732. t.Fatalf("No db update received")
  733. }
  734. if _, err := ffs.Stat("file"); err != nil {
  735. t.Errorf("Expected no error when stating file behind symlink, got %v", err)
  736. }
  737. }
  738. // Reproduces https://github.com/syncthing/syncthing/issues/6559
  739. func TestPullCtxCancel(t *testing.T) {
  740. _, f := setupSendReceiveFolder(t)
  741. pullChan := make(chan pullBlockState)
  742. finisherChan := make(chan *sharedPullerState)
  743. ctx, cancel := context.WithCancel(t.Context())
  744. go f.pullerRoutine(ctx, pullChan, finisherChan)
  745. defer close(pullChan)
  746. emptyState := func() pullBlockState {
  747. return pullBlockState{
  748. sharedPullerState: newSharedPullerState(protocol.FileInfo{}, nil, f.folderID, "", nil, nil, false, false, protocol.FileInfo{}, false, false),
  749. block: protocol.BlockInfo{},
  750. }
  751. }
  752. cancel()
  753. done := make(chan struct{})
  754. defer close(done)
  755. for i := 0; i < 2; i++ {
  756. go func() {
  757. select {
  758. case pullChan <- emptyState():
  759. case <-done:
  760. }
  761. }()
  762. select {
  763. case s := <-finisherChan:
  764. if s.failed() == nil {
  765. t.Errorf("state %v not failed", i)
  766. }
  767. case <-time.After(5 * time.Second):
  768. t.Fatalf("timed out before receiving state %v on finisherChan", i)
  769. }
  770. }
  771. }
  772. func TestPullDeleteUnscannedDir(t *testing.T) {
  773. _, f := setupSendReceiveFolder(t)
  774. ffs := f.Filesystem()
  775. dir := "foobar"
  776. must(t, ffs.MkdirAll(dir, 0o777))
  777. fi := protocol.FileInfo{
  778. Name: dir,
  779. }
  780. scanChan := make(chan string, 1)
  781. dbUpdateChan := make(chan dbUpdateJob, 1)
  782. f.deleteDir(fi, dbUpdateChan, scanChan)
  783. if _, err := ffs.Stat(dir); fs.IsNotExist(err) {
  784. t.Error("directory has been deleted")
  785. }
  786. select {
  787. case toScan := <-scanChan:
  788. if toScan != dir {
  789. t.Errorf("expected %v to be scanned, got %v", dir, toScan)
  790. }
  791. default:
  792. t.Error("nothing was scheduled for scanning")
  793. }
  794. }
  795. func TestPullCaseOnlyPerformFinish(t *testing.T) {
  796. m, f := setupSendReceiveFolder(t)
  797. ffs := f.Filesystem()
  798. name := "foo"
  799. contents := []byte("contents")
  800. writeFile(t, ffs, name, contents)
  801. must(t, f.scanSubdirs(t.Context(), nil))
  802. var cur protocol.FileInfo
  803. hasCur := false
  804. it, errFn := m.LocalFiles(f.ID, protocol.LocalDeviceID)
  805. for i := range it {
  806. if hasCur {
  807. t.Fatal("got more than one file")
  808. }
  809. cur = i
  810. hasCur = true
  811. }
  812. if err := errFn(); err != nil {
  813. t.Fatal(err)
  814. }
  815. if !hasCur {
  816. t.Fatal("file is missing")
  817. }
  818. remote := cur
  819. remote.Version = protocol.Vector{}.Update(device1.Short())
  820. remote.Name = strings.ToUpper(cur.Name)
  821. temp := fs.TempName(remote.Name)
  822. writeFile(t, ffs, temp, contents)
  823. scanChan := make(chan string, 1)
  824. dbUpdateChan := make(chan dbUpdateJob, 1)
  825. err := f.performFinish(remote, cur, hasCur, temp, dbUpdateChan, scanChan)
  826. select {
  827. case <-dbUpdateChan: // boring case sensitive filesystem
  828. return
  829. case <-scanChan:
  830. t.Error("no need to scan anything here")
  831. default:
  832. }
  833. var caseErr *fs.CaseConflictError
  834. if !errors.As(err, &caseErr) {
  835. t.Error("Expected case conflict error, got", err)
  836. }
  837. }
  838. func TestPullCaseOnlyDir(t *testing.T) {
  839. testPullCaseOnlyDirOrSymlink(t, true)
  840. }
  841. func TestPullCaseOnlySymlink(t *testing.T) {
  842. if build.IsWindows {
  843. t.Skip("symlinks not supported on windows")
  844. }
  845. testPullCaseOnlyDirOrSymlink(t, false)
  846. }
  847. func testPullCaseOnlyDirOrSymlink(t *testing.T, dir bool) {
  848. m, f := setupSendReceiveFolder(t)
  849. ffs := f.Filesystem()
  850. name := "foo"
  851. if dir {
  852. must(t, ffs.Mkdir(name, 0o777))
  853. } else {
  854. must(t, ffs.CreateSymlink("target", name))
  855. }
  856. must(t, f.scanSubdirs(t.Context(), nil))
  857. var cur protocol.FileInfo
  858. hasCur := false
  859. it, errFn := m.LocalFiles(f.ID, protocol.LocalDeviceID)
  860. for i := range it {
  861. if hasCur {
  862. t.Fatal("got more than one file")
  863. }
  864. cur = i
  865. hasCur = true
  866. }
  867. if err := errFn(); err != nil {
  868. t.Fatal(err)
  869. }
  870. if !hasCur {
  871. t.Fatal("file is missing")
  872. }
  873. scanChan := make(chan string, 1)
  874. dbUpdateChan := make(chan dbUpdateJob, 1)
  875. remote := cur
  876. remote.Version = protocol.Vector{}.Update(device1.Short())
  877. remote.Name = strings.ToUpper(cur.Name)
  878. if dir {
  879. f.handleDir(remote, dbUpdateChan, scanChan)
  880. } else {
  881. f.handleSymlink(remote, dbUpdateChan, scanChan)
  882. }
  883. select {
  884. case <-dbUpdateChan: // boring case sensitive filesystem
  885. return
  886. case <-scanChan:
  887. t.Error("no need to scan anything here")
  888. default:
  889. }
  890. if errStr, ok := f.tempPullErrors[remote.Name]; !ok {
  891. t.Error("missing error for", remote.Name)
  892. } else if !strings.Contains(errStr, "uses different upper or lowercase") {
  893. t.Error("unexpected error", errStr, "for", remote.Name)
  894. }
  895. }
  896. func TestPullTempFileCaseConflict(t *testing.T) {
  897. _, f := setupSendReceiveFolder(t)
  898. copyChan := make(chan copyBlocksState, 1)
  899. file := protocol.FileInfo{Name: "foo"}
  900. confl := "Foo"
  901. tempNameConfl := fs.TempName(confl)
  902. if fd, err := f.mtimefs.Create(tempNameConfl); err != nil {
  903. t.Fatal(err)
  904. } else {
  905. if _, err := fd.Write([]byte("data")); err != nil {
  906. t.Fatal(err)
  907. }
  908. fd.Close()
  909. }
  910. f.handleFile(t.Context(), file, copyChan)
  911. cs := <-copyChan
  912. if _, err := cs.tempFile(); err != nil {
  913. t.Error(err)
  914. } else {
  915. cs.finalClose()
  916. }
  917. }
  918. func TestPullCaseOnlyRename(t *testing.T) {
  919. m, f := setupSendReceiveFolder(t)
  920. // tempNameConfl := fs.TempName(confl)
  921. name := "foo"
  922. if fd, err := f.mtimefs.Create(name); err != nil {
  923. t.Fatal(err)
  924. } else {
  925. if _, err := fd.Write([]byte("data")); err != nil {
  926. t.Fatal(err)
  927. }
  928. fd.Close()
  929. }
  930. must(t, f.scanSubdirs(t.Context(), nil))
  931. cur, ok := m.testCurrentFolderFile(f.ID, name)
  932. if !ok {
  933. t.Fatal("file missing")
  934. }
  935. deleted := cur
  936. deleted.SetDeleted(myID.Short())
  937. confl := cur
  938. confl.Name = "Foo"
  939. confl.Version = confl.Version.Update(device1.Short())
  940. dbUpdateChan := make(chan dbUpdateJob, 2)
  941. scanChan := make(chan string, 2)
  942. if err := f.renameFile(cur, deleted, confl, dbUpdateChan, scanChan); err != nil {
  943. t.Error(err)
  944. }
  945. }
  946. func TestPullSymlinkOverExistingWindows(t *testing.T) {
  947. if !build.IsWindows {
  948. t.Skip()
  949. }
  950. m, f := setupSendReceiveFolder(t)
  951. conn := addFakeConn(m, device1, f.ID)
  952. name := "foo"
  953. if fd, err := f.mtimefs.Create(name); err != nil {
  954. t.Fatal(err)
  955. } else {
  956. if _, err := fd.Write([]byte("data")); err != nil {
  957. t.Fatal(err)
  958. }
  959. fd.Close()
  960. }
  961. must(t, f.scanSubdirs(t.Context(), nil))
  962. file, ok := m.testCurrentFolderFile(f.ID, name)
  963. if !ok {
  964. t.Fatal("file missing")
  965. }
  966. must(t, m.Index(conn, &protocol.Index{Folder: f.ID, Files: []protocol.FileInfo{{Name: name, Type: protocol.FileInfoTypeSymlink, Version: file.Version.Update(device1.Short())}}}))
  967. scanChan := make(chan string)
  968. changed, err := f.pullerIteration(t.Context(), scanChan)
  969. must(t, err)
  970. if changed != 1 {
  971. t.Error("Expected one change in pull, got", changed)
  972. }
  973. if file, ok := m.testCurrentFolderFile(f.ID, name); !ok {
  974. t.Error("symlink entry missing")
  975. } else if !file.IsUnsupported() {
  976. t.Error("symlink entry isn't marked as unsupported")
  977. }
  978. if _, err := f.mtimefs.Lstat(name); err == nil {
  979. t.Error("old file still exists on disk")
  980. } else if !fs.IsNotExist(err) {
  981. t.Error(err)
  982. }
  983. }
  984. func TestPullDeleteCaseConflict(t *testing.T) {
  985. _, f := setupSendReceiveFolder(t)
  986. name := "foo"
  987. fi := protocol.FileInfo{Name: "Foo"}
  988. dbUpdateChan := make(chan dbUpdateJob, 1)
  989. scanChan := make(chan string)
  990. if fd, err := f.mtimefs.Create(name); err != nil {
  991. t.Fatal(err)
  992. } else {
  993. if _, err := fd.Write([]byte("data")); err != nil {
  994. t.Fatal(err)
  995. }
  996. fd.Close()
  997. }
  998. f.deleteFileWithCurrent(fi, protocol.FileInfo{}, false, dbUpdateChan, scanChan)
  999. select {
  1000. case <-dbUpdateChan:
  1001. default:
  1002. t.Error("Missing db update for file")
  1003. }
  1004. f.deleteDir(fi, dbUpdateChan, scanChan)
  1005. select {
  1006. case <-dbUpdateChan:
  1007. default:
  1008. t.Error("Missing db update for dir")
  1009. }
  1010. }
  1011. func TestPullDeleteIgnoreChildDir(t *testing.T) {
  1012. _, f := setupSendReceiveFolder(t)
  1013. parent := "parent"
  1014. del := "ignored"
  1015. child := "keep"
  1016. matcher := ignore.New(f.mtimefs)
  1017. must(t, matcher.Parse(bytes.NewBufferString(fmt.Sprintf(`
  1018. !%v
  1019. (?d)%v
  1020. `, child, del)), ""))
  1021. f.ignores = matcher
  1022. must(t, f.mtimefs.Mkdir(parent, 0o777))
  1023. must(t, f.mtimefs.Mkdir(filepath.Join(parent, del), 0o777))
  1024. must(t, f.mtimefs.Mkdir(filepath.Join(parent, del, child), 0o777))
  1025. scanChan := make(chan string, 2)
  1026. err := f.deleteDirOnDisk(parent, scanChan)
  1027. if err == nil {
  1028. t.Error("no error")
  1029. }
  1030. }
  1031. func cleanupSharedPullerState(s *sharedPullerState) {
  1032. s.mut.Lock()
  1033. defer s.mut.Unlock()
  1034. if s.writer == nil {
  1035. return
  1036. }
  1037. s.writer.mut.Lock()
  1038. s.writer.fd.Close()
  1039. s.writer.mut.Unlock()
  1040. }
  1041. func startCopier(ctx context.Context, f *sendReceiveFolder, pullChan chan<- pullBlockState, finisherChan chan<- *sharedPullerState) (chan copyBlocksState, *sync.WaitGroup) {
  1042. copyChan := make(chan copyBlocksState)
  1043. wg := new(sync.WaitGroup)
  1044. wg.Add(1)
  1045. go func() {
  1046. f.copierRoutine(ctx, copyChan, pullChan, finisherChan)
  1047. wg.Done()
  1048. }()
  1049. return copyChan, wg
  1050. }