mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-07 04:27:49 +01:00
remote_storage: bound the sync offset write like the read
The offset write runs in the subscription's checkpoint callback and, now, when a subscription ends; a filer that never answers it stalled either. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
7be919ded1
commit
f888ac6b8f
1 file changed
+8
-3
@@ -24,11 +24,14 @@ func SyncOffsetKey(dir string) []byte {
|
||||
return syncKey
|
||||
}
|
||||
|
||||
// syncOffsetTimeout bounds an offset read or write: one that never answers
|
||||
// would hold the daemon's start, or its event stream behind the checkpoint.
|
||||
const syncOffsetTimeout = 30 * time.Second
|
||||
|
||||
func GetSyncOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, dir string) (lastOffsetTsNs int64, readErr error) {
|
||||
|
||||
readErr = pb.WithFilerClient(false, 0, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
|
||||
// bounded: a read that never answers would hold the daemon's start
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), syncOffsetTimeout)
|
||||
defer cancel()
|
||||
resp, err := client.KvGet(ctx, &filer_pb.KvGetRequest{Key: SyncOffsetKey(dir)})
|
||||
if err != nil {
|
||||
@@ -58,7 +61,9 @@ func SetSyncOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, dir s
|
||||
valueBuf := make([]byte, 8)
|
||||
util.Uint64toBytes(valueBuf, uint64(offsetTsNs))
|
||||
|
||||
resp, err := client.KvPut(context.Background(), &filer_pb.KvPutRequest{
|
||||
ctx, cancel := context.WithTimeout(context.Background(), syncOffsetTimeout)
|
||||
defer cancel()
|
||||
resp, err := client.KvPut(ctx, &filer_pb.KvPutRequest{
|
||||
Key: SyncOffsetKey(dir),
|
||||
Value: valueBuf,
|
||||
})
|
||||
|
||||
Reference in new issue
Block a user