337
338
// TestConcurrentlyCreateAndAppendBranches test
339
>
func (s *HistoryV2PersistenceSuite) TestConcurrentlyCreateAndAppendBranches() {
history_v2_persistence.go
340
>
treeID := uuid.NewString()
341
>
wg := sync.WaitGroup{}
342
>
concurrency := 1
343
>
m := &sync.Map{}
344
>
345
>
// test create new branch along with appending new nodes
346
>
for i := range concurrency {
347
>
wg.Add(1)
348
>
go func(idx int) {
349
>
defer wg.Done()
350
>
bi, err := s.newHistoryBranch(treeID)
351
>
s.Nil(err)
352
>
historyW := &historypb.History{}
353
>
m.Store(idx, bi)
354
>
355
>
events := s.genRandomEvents([]int64{1, 2, 3}, 1)
356
>
err = s.appendNewBranchAndFirstNode(bi, events, 1, "branchInfo")
357
>
s.Nil(err)
358
>
historyW.Events = events
359
>
360
>
events = s.genRandomEvents([]int64{4}, 1)
361
>
err = s.appendNewNode(bi, events, 2)
362
>
s.Nil(err)
363
>
historyW.Events = append(historyW.Events, events...)
364
>
365
>
events = s.genRandomEvents([]int64{5, 6, 7, 8}, 1)
366
>
err = s.appendNewNode(bi, events, 3)
367
>
s.Nil(err)
368
>
historyW.Events = append(historyW.Events, events...)
369
>
370
>
events = s.genRandomEvents([]int64{9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20}, 1)
371
>
err = s.appendNewNode(bi, events, 4000)
372
>
s.Nil(err)
373
>
historyW.Events = append(historyW.Events, events...)
374
>
375
>
// read branch to verify
376
>
historyR := &historypb.History{}
377
>
events = s.read(bi, 1, 21)
378
>
s.Equal(20, len(events))
379
>
historyR.Events = events
380
>
381
>
s.ProtoEqual(historyW, historyR)
382
>
}(i)
383
}
384
386
>
branches := s.descTree(treeID)
387
>
s.Equal(concurrency, len(branches))
388
>
389
>
wg = sync.WaitGroup{}
390
>
// test appending nodes(override and new nodes) on each branch concurrently
391
>
for i := range concurrency {
392
>
wg.Add(1)
393
>
go func(idx int) {
394
>
defer wg.Done()
395
>
396
>
branch := s.getBranchByKey(m, idx)
397
>
398
>
// override with smaller txn_id
399
>
events := s.genRandomEvents([]int64{5}, 1)
400
>
err := s.appendNewNode(branch, events, 0)
401
>
s.Nil(err)
402
>
// it shouldn't change anything
403
>
events = s.read(branch, 1, 25)
404
>
s.Equal(20, len(events))
405
>
406
>
// override with greatest txn_id
407
>
events = s.genRandomEvents([]int64{5}, 1)
408
>
err = s.appendNewNode(branch, events, 3000)
409
>
s.Nil(err)
410
>
411
>
// read to verify override success, at this point history is corrupted, missing 6/7/8, so we should only see 5 events
412
>
events = s.read(branch, 1, 6)
413
>
s.Equal(5, len(events))
414
>
_, err = s.readWithError(branch, 1, 25)
415
>
_, ok := err.(*serviceerror.DataLoss)
416
>
s.Equal(true, ok)
417
>
418
>
// override with even larger txn_id and same version
419
>
events = s.genRandomEvents([]int64{5, 6}, 1)
420
>
err = s.appendNewNode(branch, events, 3001)
421
>
s.Nil(err)
422
>
423
>
// read to verify override success, at this point history is corrupted, missing 7/8, so we should only see 6 events
424
>
events = s.read(branch, 1, 7)
425
>
s.Equal(6, len(events))
426
>
_, err = s.readWithError(branch, 1, 25)
427
>
_, ok = err.(*serviceerror.DataLoss)
428
>
s.Equal(true, ok)
429
>
430
>
// override more with larger txn_id, this would fix the corrupted hole so that we cna get 20 events again
431
>
events = s.genRandomEvents([]int64{7, 8}, 1)
432
>
err = s.appendNewNode(branch, events, 3002)
433
>
s.Nil(err)
434
>
435
>
// read to verify override
436
>
events = s.read(branch, 1, 25)
437
>
s.Equal(20, len(events))
438
>
events = s.genRandomEvents([]int64{9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23}, 1)
439
>
err = s.appendNewNode(branch, events, 4001)
440
>
s.Nil(err)
441
>
events = s.read(branch, 1, 25)
442
>
s.Equal(23, len(events))
443
>
}(i)
444
}
445
447
>
// Finally lets clean up all branches
448
>
m.Range(func(k, v any) bool {
449
>
br := v.([]byte)
450
>
// delete old branches along with create new branches
451
>
err := s.deleteHistoryBranch(br)
452
>
s.Nil(err)
453
>
return true
454
>
})
455
457
>
s.Equal(0, len(branches))
458
}
459