Skip to content

Commit e4bd1ad

Browse files
psviderskipsviderski
committed
feat(store): improve waiting for store replication to skip unavailable members
1 parent 0e53f50 commit e4bd1ad

7 files changed

Lines changed: 76 additions & 26 deletions

File tree

‎api/pb/machine.proto‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ service Machine {
2020
// InspectMachine retrieves detailed information about the machine. Supports broadcasting to multiple machines.
2121
rpc InspectMachine(google.protobuf.Empty) returns (InspectMachineResponse);
2222
// WaitForStoreVersion waits until the cluster store on this machine has reached each requested actor version, with
23-
// no known missing or pending transactions through those versions.
23+
// no known missing or pending transactions through those versions from active members.
2424
// Corrosion may satisfy a version by applying its surviving changes or by marking it complete because its changes
2525
// have been superseded.
2626
//

‎api/pb/machine_grpc.pb.go‎

Lines changed: 2 additions & 2 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎internal/corrosion/admin.go‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@ import (
1313
"time"
1414
)
1515

16+
const adminCommandTimeout = 5 * time.Second
17+
1618
// AdminClient is a client for the Corrosion admin API.
1719
type AdminClient struct {
1820
sockPath string
@@ -32,10 +34,15 @@ type Response struct {
3234
// The channel will be closed after sending the last or error response. The caller must read from the channel until
3335
// it is closed.
3436
func (c *AdminClient) SendCommand(cmd []byte) (<-chan Response, error) {
35-
conn, err := net.Dial("unix", c.sockPath)
37+
// TODO: Accept contexts in admin methods so callers can cancel requests and set their deadlines.
38+
conn, err := net.DialTimeout("unix", c.sockPath, adminCommandTimeout)
3639
if err != nil {
3740
return nil, fmt.Errorf("connect to admin socket: %w", err)
3841
}
42+
if err = conn.SetDeadline(time.Now().Add(adminCommandTimeout)); err != nil {
43+
conn.Close()
44+
return nil, fmt.Errorf("set admin connection deadline: %w", err)
45+
}
3946

4047
if _, err = conn.Write(encodeFrame(cmd)); err != nil {
4148
conn.Close()

‎internal/machine/machine.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -262,11 +262,11 @@ func NewMachine(config *Config) (*Machine, error) {
262262
if err != nil {
263263
return nil, fmt.Errorf("create corrosion API client: %w", err)
264264
}
265-
corroStore := store.New(corro)
266265
corroAdmin, err := corrosion.NewAdminClient(config.CorrosionAdminSockPath)
267266
if err != nil {
268267
return nil, fmt.Errorf("create corrosion admin client: %w", err)
269268
}
269+
corroStore := store.New(corro, corroAdmin)
270270

271271
initialised := make(chan struct{})
272272
clusterReady := make(chan struct{})

‎internal/machine/store/replication.go‎

Lines changed: 59 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,16 @@ import (
88
"time"
99

1010
"github.com/google/uuid"
11+
"github.com/psviderski/uncloud/internal/corrosion"
1112
"github.com/psviderski/uncloud/pkg/api"
1213
)
1314

1415
// ErrInvalidStoreVersion indicates an invalid actor UUID in a store version vector.
1516
var ErrInvalidStoreVersion = errors.New("invalid store version")
1617

1718
// WaitForVersion waits until the local store has reached each actor's minimum version in minVersion, with no known
18-
// missing or pending transactions through those versions. Corrosion may satisfy a version by applying its surviving
19-
// changes or by marking it complete because its changes have been superseded.
19+
// missing or pending transactions through those versions from active members. Corrosion may satisfy a version
20+
// by applying its surviving changes or by marking it complete because its changes have been superseded.
2021
//
2122
// Waiting normally makes the captured data available locally. However, if another write replaces some of that data
2223
// before it arrives, Corrosion can complete the older version without transferring the replaced data. If the
@@ -53,28 +54,64 @@ func (s *Store) WaitForVersion(ctx context.Context, minVersion api.StoreVersion)
5354
return nil
5455
}
5556

56-
// Check all bookkeeping in one SQLite snapshot. Raw buffered rows can remain after application,
57-
// but sequence bookkeeping is removed in the same transaction that applies or clears a version.
58-
query := `WITH min_version(actor_id, version) AS (VALUES ` + strings.Join(placeholders, ", ") + `)
57+
queryPrefix := `WITH min_version(actor_id, version) AS (VALUES ` + strings.Join(placeholders, ", ") + `)
5958
SELECT NOT EXISTS (
6059
SELECT 1 FROM min_version
6160
LEFT JOIN crsql_db_versions AS current ON current.site_id = min_version.actor_id
62-
WHERE COALESCE(current.db_version, 0) < min_version.version
63-
OR EXISTS (
64-
SELECT 1 FROM __corro_bookkeeping_gaps
65-
WHERE actor_id = min_version.actor_id AND start <= min_version.version
66-
)
67-
OR EXISTS (
68-
SELECT 1 FROM __corro_seq_bookkeeping
69-
WHERE site_id = min_version.actor_id AND db_version <= min_version.version
70-
)
71-
)`
61+
WHERE COALESCE(current.db_version, 0) < min_version.version`
62+
63+
versionsReached := false
7264

7365
ticker := time.NewTicker(100 * time.Millisecond)
7466
defer ticker.Stop()
7567

7668
for {
77-
rows, err := s.corro.QueryContext(ctx, query, args...)
69+
query := queryPrefix
70+
queryArgs := args
71+
// Check for gaps and pending transactions only after the requested versions are reached.
72+
if versionsReached {
73+
states, err := s.corroAdmin.ClusterMembershipStates(true)
74+
if err != nil {
75+
return fmt.Errorf("get cluster membership for store replication: %w", err)
76+
}
77+
78+
activePlaceholders := make([]string, 0, len(states))
79+
queryArgs = make([]any, len(args), len(args)+len(states))
80+
copy(queryArgs, args)
81+
for _, state := range states {
82+
if state.State != corrosion.MembershipStateAlive && state.State != corrosion.MembershipStateSuspect {
83+
continue
84+
}
85+
actorID, err := uuid.Parse(state.ID)
86+
if err != nil {
87+
return fmt.Errorf("parse cluster member actor '%s': %w", state.ID, err)
88+
}
89+
activePlaceholders = append(activePlaceholders, "?")
90+
queryArgs = append(queryArgs, [16]byte(actorID))
91+
}
92+
if len(activePlaceholders) > 0 {
93+
// Check active members' gaps and pending sequences in the same query for simplicity.
94+
// Buffered rows can remain in __corro_buffered_changes after application, but sequence bookkeeping
95+
// is removed when a version completes.
96+
query += `
97+
OR (
98+
min_version.actor_id IN (` + strings.Join(activePlaceholders, ", ") + `)
99+
AND (
100+
EXISTS (
101+
SELECT 1 FROM __corro_bookkeeping_gaps
102+
WHERE actor_id = min_version.actor_id AND start <= min_version.version
103+
)
104+
OR EXISTS (
105+
SELECT 1 FROM __corro_seq_bookkeeping
106+
WHERE site_id = min_version.actor_id AND db_version <= min_version.version
107+
)
108+
)
109+
)`
110+
}
111+
}
112+
query += ")"
113+
114+
rows, err := s.corro.QueryContext(ctx, query, queryArgs...)
78115
if err != nil {
79116
return fmt.Errorf("check store replication: %w", err)
80117
}
@@ -84,6 +121,7 @@ func (s *Store) WaitForVersion(ctx context.Context, minVersion api.StoreVersion)
84121
}
85122
return errors.New("check store replication: store version query returned no rows")
86123
}
124+
87125
var reached int
88126
if err = rows.Scan(&reached); err != nil {
89127
rows.Close()
@@ -95,7 +133,11 @@ func (s *Store) WaitForVersion(ctx context.Context, minVersion api.StoreVersion)
95133
return fmt.Errorf("check store replication: %w", err)
96134
}
97135
if reached == 1 {
98-
return ctx.Err()
136+
if versionsReached {
137+
return nil
138+
}
139+
versionsReached = true
140+
continue
99141
}
100142

101143
select {

‎internal/machine/store/store.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,12 @@ var (
2424

2525
// Store is a cluster store backed by a distributed Corrosion database.
2626
type Store struct {
27-
corro *corrosion.APIClient
27+
corro *corrosion.APIClient
28+
corroAdmin *corrosion.AdminClient
2829
}
2930

30-
func New(corro *corrosion.APIClient) *Store {
31-
return &Store{corro: corro}
31+
func New(corro *corrosion.APIClient, corroAdmin *corrosion.AdminClient) *Store {
32+
return &Store{corro: corro, corroAdmin: corroAdmin}
3233
}
3334

3435
// Get retrieves an unnamespaced legacy value.

‎pkg/client/machine.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,7 @@ func (cli *Client) WaitClusterReady(ctx context.Context, timeout time.Duration)
143143
}
144144

145145
// WaitForStoreVersion waits until the cluster store on the target machine has reached each requested actor version
146-
// in minVersion, with no known missing or pending transactions through those versions.
146+
// in minVersion, with no known missing or pending transactions through those versions from active members.
147147
// The context controls cancellation and the deadline. An empty minVersion requires no replication.
148148
// This method observes replication without initiating synchronisation.
149149
//

0 commit comments

Comments
 (0)