Skip to content

Commit 5dfb480

Browse files
VioletQwQ-0VioletQwQ-0
committed
fix(incrservice): bound and isolate offset resets
1 parent 0018047 commit 5dfb480

5 files changed

Lines changed: 150 additions & 39 deletions

File tree

‎pkg/common/moerr/cause.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -104,10 +104,11 @@ var (
104104
CauseAllocateTasks = NewInternalError(context.Background(), "allocateTask")
105105
CauseTruncateTasks = NewInternalError(context.Background(), "truncateTasks")
106106
//pkg/incrservice
107-
CauseDoAllocate = NewInternalError(context.Background(), "doAllocate")
108-
CauseDoUpdate = NewInternalError(context.Background(), "doUpdate")
109-
CauseDestroyTables = NewInternalError(context.Background(), "destroyTables")
110-
CauseAllocate = NewInternalError(context.Background(), "allocate")
107+
CauseDoAllocate = NewInternalError(context.Background(), "doAllocate")
108+
CauseDoUpdate = NewInternalError(context.Background(), "doUpdate")
109+
CauseDoForceSetOffset = NewInternalError(context.Background(), "doForceSetOffset")
110+
CauseDestroyTables = NewInternalError(context.Background(), "destroyTables")
111+
CauseAllocate = NewInternalError(context.Background(), "allocate")
111112
//pkg/lockservice
112113
CauseCleanCommitState = NewInternalError(context.Background(), "cleanCommitState")
113114
CauseValidateService = NewInternalError(context.Background(), "validateService")

‎pkg/common/moerr/cause_test.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,7 @@ var causeArray = []error{
9898

9999
CauseDoAllocate,
100100
CauseDoUpdate,
101+
CauseDoForceSetOffset,
101102
CauseDestroyTables,
102103
CauseAllocate,
103104

‎pkg/incrservice/allocator.go‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,8 @@ import (
3131
)
3232

3333
const (
34-
defaultAllocateTimeout = time.Minute * 3
34+
defaultAllocateTimeout = time.Minute * 3
35+
defaultForceSetOffsetTimeout = time.Second * 10
3536
)
3637

3738
type allocator struct {
@@ -248,12 +249,19 @@ func (a *allocator) run(ctx context.Context) {
248249
}
249250

250251
func (a *allocator) doForceSetOffset(act action) {
251-
ctx := defines.AttachAccountId(act.ctx, act.accountID)
252+
baseCtx := act.ctx
253+
if baseCtx == nil {
254+
baseCtx = context.Background()
255+
}
256+
ctx := defines.AttachAccountId(baseCtx, act.accountID)
257+
ctx, cancel := context.WithTimeoutCause(ctx, defaultForceSetOffsetTimeout, moerr.CauseDoForceSetOffset)
258+
defer cancel()
252259
if err := context.Cause(ctx); err != nil {
253260
act.applyUpdate(err)
254261
return
255262
}
256263
err := a.store.ForceSetOffset(ctx, act.tableID, act.col, act.minValue, act.txnOp)
264+
err = moerr.AttachCause(ctx, err)
257265
act.applyUpdate(err)
258266
}
259267

‎pkg/incrservice/service_test.go‎

Lines changed: 81 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,26 @@ type blockingAllocateStore struct {
7373
release chan struct{}
7474
}
7575

76+
type deadlineCheckingForceSetOffsetStore struct {
77+
IncrValueStore
78+
deadline chan time.Time
79+
}
80+
81+
func (s *deadlineCheckingForceSetOffsetStore) ForceSetOffset(
82+
ctx context.Context,
83+
tableID uint64,
84+
colName string,
85+
offset uint64,
86+
txnOp client.TxnOperator,
87+
) error {
88+
deadline, ok := ctx.Deadline()
89+
if !ok {
90+
return errors.New("ForceSetOffset context has no deadline")
91+
}
92+
s.deadline <- deadline
93+
return s.IncrValueStore.ForceSetOffset(ctx, tableID, colName, offset, txnOp)
94+
}
95+
7696
func (s *blockingAllocateStore) blockNext() (<-chan struct{}, func()) {
7797
s.mu.Lock()
7898
defer s.mu.Unlock()
@@ -501,9 +521,16 @@ func TestMemStoreSetOffsetReturnsError(t *testing.T) {
501521
require.Error(t, store.SetOffset(ctx, 0, "missing_col", 77, nil))
502522

503523
op := ops[0]
524+
require.Error(t, store.SetOffset(ctx, 1, def[0].ColName, 77, op))
525+
store.Lock()
526+
_, exists := store.uncommitted[string(op.Txn().ID)]
527+
store.Unlock()
528+
require.False(t, exists)
529+
504530
require.NoError(t, store.SetOffset(ctx, 0, def[0].ColName, 88, op))
505531
store.Lock()
506-
require.Equal(t, uint64(88), store.caches[0][0].Offset)
532+
require.Equal(t, uint64(0), store.caches[0][0].Offset)
533+
require.Equal(t, uint64(88), store.uncommitted[string(op.Txn().ID)][0][0].Offset)
507534
store.Unlock()
508535
})
509536
}
@@ -736,9 +763,6 @@ func TestSetOffsetRollbackKeepsCommittedOffsetAndSafelyRebuildsCache(t *testing.
736763

737764
alterTxn, err := tc.New(ctx, timestamp.Timestamp{})
738765
require.NoError(t, err)
739-
// The mem store models the SQL transaction's private view by copying the
740-
// committed auto-increment row into this transaction.
741-
require.NoError(t, store.Create(ctx, 0, committedCols, alterTxn))
742766
require.NoError(t, s.SetOffset(ctx, 0, def[0].ColName, 50, alterTxn))
743767

744768
store.Lock()
@@ -780,7 +804,6 @@ func TestSetOffsetTransactionUsesPendingOffsetForInsert(t *testing.T) {
780804

781805
alterTxn, err := tc.New(ctx, timestamp.Timestamp{})
782806
require.NoError(t, err)
783-
require.NoError(t, store.Create(ctx, 0, def, alterTxn))
784807
require.NoError(t, s.SetOffset(ctx, 0, def[0].ColName, 999, alterTxn))
785808

786809
input := newTestVector[uint64](1, types.New(types.T_uint64, 0, 0), nil, nil)
@@ -883,7 +906,6 @@ func TestSetOffsetWithoutInsertDoesNotReservePrivateRange(t *testing.T) {
883906

884907
alterTxn, err := tc.New(ctx, timestamp.Timestamp{})
885908
require.NoError(t, err)
886-
require.NoError(t, store.Create(ctx, 0, def, alterTxn))
887909
require.NoError(t, s.SetOffset(ctx, 0, def[0].ColName, 999, alterTxn))
888910

889911
store.Lock()
@@ -1431,8 +1453,6 @@ func TestTwoServicesKeepStaleRangesAcrossTransactionalReset(t *testing.T) {
14311453

14321454
alterTxn, err := tc.New(ctx, timestamp.Timestamp{})
14331455
require.NoError(t, err)
1434-
// The memory store models the ALTER transaction's private metadata row.
1435-
require.NoError(t, store.Create(ctx, 0, def, alterTxn))
14361456
require.NoError(t, cn1.SetOffset(ctx, 0, def[0].ColName, effectiveOffset, alterTxn))
14371457
require.NoError(t, alterTxn.Commit(ctx))
14381458

@@ -1715,6 +1735,23 @@ func TestCanceledSetOffsetDoesNotRunQueuedForceUpdate(t *testing.T) {
17151735
})
17161736
}
17171737

1738+
func TestForceSetOffsetUsesBoundedContext(t *testing.T) {
1739+
ctx := defines.AttachAccountId(context.Background(), catalog.System_Account)
1740+
store := &deadlineCheckingForceSetOffsetStore{
1741+
IncrValueStore: NewMemStore(),
1742+
deadline: make(chan time.Time, 1),
1743+
}
1744+
require.NoError(t, store.Create(ctx, 0, newTestTableDef(1), nil))
1745+
allocator := newValueAllocator("", store).(*allocator)
1746+
defer allocator.close()
1747+
1748+
require.NoError(t, allocator.forceSetOffset(ctx, 0, "auto_0", 99, nil))
1749+
deadline := <-store.deadline
1750+
remaining := time.Until(deadline)
1751+
require.Positive(t, remaining)
1752+
require.LessOrEqual(t, remaining, defaultForceSetOffsetTimeout)
1753+
}
1754+
17181755
func TestRetiredTableCacheCannotQueueAllocation(t *testing.T) {
17191756
allocator := &countingAllocator{}
17201757
col := &columnCache{
@@ -1852,6 +1889,42 @@ func TestMemStoreForceSetOffsetLowerThanCurrent(t *testing.T) {
18521889
})
18531890
}
18541891

1892+
func TestMemStoreForceSetOffsetCreatesTransactionPrivateState(t *testing.T) {
1893+
client.RunTxnTests(func(tc client.TxnClient, _ rpc.TxnSender) {
1894+
ctx, cancel := context.WithTimeout(defines.AttachAccountId(context.Background(), catalog.System_Account), 10*time.Second)
1895+
defer cancel()
1896+
store := NewMemStore().(*memStore)
1897+
def := newTestTableDef(1)
1898+
createTxn, err := tc.New(ctx, timestamp.Timestamp{})
1899+
require.NoError(t, err)
1900+
require.NoError(t, store.Create(ctx, 0, def, createTxn))
1901+
require.NoError(t, createTxn.Commit(ctx))
1902+
require.NoError(t, store.ForceSetOffset(ctx, 0, def[0].ColName, 10, nil))
1903+
1904+
rollbackTxn, err := tc.New(ctx, timestamp.Timestamp{})
1905+
require.NoError(t, err)
1906+
require.NoError(t, store.ForceSetOffset(ctx, 0, def[0].ColName, 99, rollbackTxn))
1907+
store.Lock()
1908+
require.Equal(t, uint64(10), store.caches[0][0].Offset)
1909+
require.Equal(t, uint64(99), store.uncommitted[string(rollbackTxn.Txn().ID)][0][0].Offset)
1910+
store.Unlock()
1911+
require.NoError(t, rollbackTxn.Rollback(ctx))
1912+
store.Lock()
1913+
require.Equal(t, uint64(10), store.caches[0][0].Offset)
1914+
_, exists := store.uncommitted[string(rollbackTxn.Txn().ID)]
1915+
store.Unlock()
1916+
require.False(t, exists)
1917+
1918+
commitTxn, err := tc.New(ctx, timestamp.Timestamp{})
1919+
require.NoError(t, err)
1920+
require.NoError(t, store.ForceSetOffset(ctx, 0, def[0].ColName, 77, commitTxn))
1921+
require.NoError(t, commitTxn.Commit(ctx))
1922+
store.Lock()
1923+
require.Equal(t, uint64(77), store.caches[0][0].Offset)
1924+
store.Unlock()
1925+
})
1926+
}
1927+
18551928
func runServiceTests(
18561929
t *testing.T,
18571930
n int,

‎pkg/incrservice/store_mem.go‎

Lines changed: 53 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -57,23 +57,7 @@ func (s *memStore) Create(
5757
defer s.Unlock()
5858
m := s.caches
5959
if txnOp != nil {
60-
m = make(map[uint64][]AutoColumn)
61-
s.uncommitted[string(txnOp.Txn().ID)] = m
62-
txnOp.AppendEventCallback(
63-
client.ClosedEvent,
64-
client.NewTxnEventCallback(
65-
func(ctx context.Context, txnOp client.TxnOperator, event client.TxnEvent, v any) error {
66-
txnMeta := event.Txn
67-
s.Lock()
68-
defer s.Unlock()
69-
delete(s.uncommitted, string(txnMeta.ID))
70-
if txnMeta.Status == txn.TxnStatus_Committed {
71-
for k, v := range m {
72-
s.caches[k] = v
73-
}
74-
}
75-
return nil
76-
}))
60+
m = s.getOrCreateTxnMapLocked(txnOp)
7761
}
7862

7963
caches := m[tableID]
@@ -93,6 +77,50 @@ func (s *memStore) Create(
9377
return nil
9478
}
9579

80+
func (s *memStore) getOrCreateTxnMapLocked(txnOp client.TxnOperator) map[uint64][]AutoColumn {
81+
txnKey := string(txnOp.Txn().ID)
82+
if m, ok := s.uncommitted[txnKey]; ok {
83+
return m
84+
}
85+
m := make(map[uint64][]AutoColumn)
86+
s.uncommitted[txnKey] = m
87+
txnOp.AppendEventCallback(
88+
client.ClosedEvent,
89+
client.NewTxnEventCallback(
90+
func(ctx context.Context, txnOp client.TxnOperator, event client.TxnEvent, v any) error {
91+
txnMeta := event.Txn
92+
s.Lock()
93+
defer s.Unlock()
94+
delete(s.uncommitted, string(txnMeta.ID))
95+
if txnMeta.Status == txn.TxnStatus_Committed {
96+
for k, v := range m {
97+
s.caches[k] = v
98+
}
99+
}
100+
return nil
101+
}))
102+
return m
103+
}
104+
105+
func (s *memStore) getOrCloneTxnTableLocked(
106+
ctx context.Context,
107+
tableID uint64,
108+
txnOp client.TxnOperator,
109+
) (map[uint64][]AutoColumn, error) {
110+
if m, ok := s.uncommitted[string(txnOp.Txn().ID)]; ok {
111+
if _, exists := m[tableID]; exists {
112+
return m, nil
113+
}
114+
}
115+
cols, ok := s.caches[tableID]
116+
if !ok {
117+
return nil, moerr.NewInternalErrorf(ctx, "incrservice: table %d not found in memStore", tableID)
118+
}
119+
m := s.getOrCreateTxnMapLocked(txnOp)
120+
m[tableID] = append([]AutoColumn(nil), cols...)
121+
return m, nil
122+
}
123+
96124
func (s *memStore) GetColumns(
97125
ctx context.Context,
98126
tableID uint64,
@@ -187,10 +215,10 @@ func (s *memStore) SetOffset(
187215
defer s.Unlock()
188216
m := s.caches
189217
if txnOp != nil {
190-
if um, ok := s.uncommitted[string(txnOp.Txn().ID)]; ok {
191-
if _, exists := um[tableID]; exists {
192-
m = um
193-
}
218+
var err error
219+
m, err = s.getOrCloneTxnTableLocked(ctx, tableID, txnOp)
220+
if err != nil {
221+
return err
194222
}
195223
}
196224
cols, ok := m[tableID]
@@ -222,10 +250,10 @@ func (s *memStore) ForceSetOffset(
222250
defer s.Unlock()
223251
m := s.caches
224252
if txnOp != nil {
225-
if um, ok := s.uncommitted[string(txnOp.Txn().ID)]; ok {
226-
if _, exists := um[tableID]; exists {
227-
m = um
228-
}
253+
var err error
254+
m, err = s.getOrCloneTxnTableLocked(ctx, tableID, txnOp)
255+
if err != nil {
256+
return err
229257
}
230258
}
231259
cols, ok := m[tableID]

0 commit comments

Comments
 (0)