folder_sendrecv_test.go 36 KB

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