1919
2020import java .util .Objects ;
2121import java .util .Locale ;
22+ import java .nio .file .FileSystems ;
23+ import java .nio .file .Path ;
24+ import java .nio .file .Paths ;
25+ import java .nio .file .StandardWatchEventKinds ;
26+ import java .nio .file .WatchEvent ;
27+ import java .nio .file .WatchKey ;
28+ import java .nio .file .WatchService ;
2229import java .util .concurrent .Executors ;
2330import java .util .concurrent .ScheduledExecutorService ;
2431import java .util .concurrent .ScheduledFuture ;
@@ -52,10 +59,10 @@ public class RocksDBCloudSession extends RocksDBSession {
5259 private static final String KEY_SYNC_INCREMENTAL_LEGACY =
5360 "rocksdb.cloud_sync_incremental" ;
5461
55- private static final String KEY_CLOUD_FIRST_MODE = "rocksdb.cloud.cloud_first_mode " ;
56- private static final String KEY_CLOUD_FIRST_MODE_LEGACY = "rocksdb.cloud_cloud_first_mode " ;
62+ private static final String KEY_SYNCHRONOUS_SST_UPLOAD_MODE = "rocksdb.cloud.synchronous_sst_upload_mode " ;
63+ private static final String KEY_SYNCHRONOUS_SST_UPLOAD_MODE_LEGACY = "rocksdb.cloud_synchronous_sst_upload_mode " ;
5764
58- private static final String KEY_SYNC_RETRY_MAX = "rocksdb.cloud.sync_retry_max" ;
65+ private static final String KEY_SYNC_RETRY_MAX = "rocksdb.cloud.sync_retry_max" ;
5966 private static final String KEY_SYNC_RETRY_MAX_LEGACY = "rocksdb.cloud_sync_retry_max" ;
6067
6168 private static final String KEY_SYNC_RETRY_BACKOFF_MS = "rocksdb.cloud.sync_retry_backoff_ms" ;
@@ -71,27 +78,30 @@ public class RocksDBCloudSession extends RocksDBSession {
7178 return t ;
7279 });
7380
74- private final CloudStorageClient storageClient ;
75- private final String bucket ;
76- private final String objectPrefix ;
77- private final int syncIntervalSeconds ;
78- private final boolean syncIncremental ;
79- private final boolean cloudFirstMode ;
80- private final int syncRetryMax ;
81- private final int syncRetryBackoffMs ;
82- private final int syncRetryMaxBackoffMs ;
81+ private final CloudStorageClient storageClient ;
82+ private final String bucket ;
83+ private final String objectPrefix ;
84+ private final int syncIntervalSeconds ;
85+ private final boolean syncIncremental ;
86+ private final boolean synchronousSstUploadMode ;
87+ private final int syncRetryMax ;
88+ private final int syncRetryBackoffMs ;
89+ private final int syncRetryMaxBackoffMs ;
8390
8491 private final AtomicBoolean syncInProgress = new AtomicBoolean (false );
8592 private final AtomicBoolean hydrationInProgress = new AtomicBoolean (false );
93+ private final AtomicBoolean sstSyncQueued = new AtomicBoolean (false );
8694
8795 private ScheduledFuture <?> periodicSyncFuture ;
96+ private WatchService sstWatchService ;
97+ private Thread sstWatchThread ;
8898
8999 public RocksDBCloudSession (HugeConfig hugeConfig , String dbDataPath ,
90100 String graphName , long version ) {
91101 super (hugeConfig , dbDataPath , graphName , version );
92102
93103 boolean cloudEnabled = getBoolean (hugeConfig , "rocksdb.cloud.enabled" ,
94- "rocksdb.cloud_enabled" , true );
104+ "rocksdb.cloud_enabled" );
95105 if (!cloudEnabled ) {
96106 log .warn ("RocksDBCloudSession is initialized while cloud sync is disabled for graph {}" ,
97107 graphName );
@@ -115,27 +125,28 @@ public RocksDBCloudSession(HugeConfig hugeConfig, String dbDataPath,
115125 KEY_PREFIX_LEGACY );
116126 this .objectPrefix = normalizedPrefix (basePrefix , graphName );
117127
118- this .syncIntervalSeconds = getInt (hugeConfig , KEY_SYNC_INTERVAL ,
119- KEY_SYNC_INTERVAL_LEGACY , 60 );
120- this .syncIncremental = getBoolean (hugeConfig , KEY_SYNC_INCREMENTAL ,
121- KEY_SYNC_INCREMENTAL_LEGACY , true );
122- this .cloudFirstMode = getBoolean (hugeConfig , KEY_CLOUD_FIRST_MODE ,
123- KEY_CLOUD_FIRST_MODE_LEGACY ,
124- false );
125- this .syncRetryMax = getInt (hugeConfig , KEY_SYNC_RETRY_MAX ,
126- KEY_SYNC_RETRY_MAX_LEGACY , 100 );
128+ this .syncIntervalSeconds = getInt (hugeConfig , KEY_SYNC_INTERVAL ,
129+ KEY_SYNC_INTERVAL_LEGACY , 60 );
130+ this .syncIncremental = getBoolean (hugeConfig , KEY_SYNC_INCREMENTAL ,
131+ KEY_SYNC_INCREMENTAL_LEGACY );
132+ this .synchronousSstUploadMode = getBoolean (hugeConfig , KEY_SYNCHRONOUS_SST_UPLOAD_MODE ,
133+ KEY_SYNCHRONOUS_SST_UPLOAD_MODE_LEGACY );
134+ this .syncRetryMax = getInt (hugeConfig , KEY_SYNC_RETRY_MAX ,
135+ KEY_SYNC_RETRY_MAX_LEGACY , 100 );
127136 this .syncRetryBackoffMs = getInt (hugeConfig , KEY_SYNC_RETRY_BACKOFF_MS ,
128137 KEY_SYNC_RETRY_BACKOFF_MS_LEGACY , 10 );
129138 this .syncRetryMaxBackoffMs = getInt (hugeConfig , KEY_SYNC_RETRY_MAX_BACKOFF_MS ,
130139 KEY_SYNC_RETRY_MAX_BACKOFF_MS_LEGACY , 1000 );
131140
132- startPeriodicSync ();
133- log .info ("RocksDB cloud enabled for graph {}: {}://{}/{}, interval={}s, " +
134- "incremental={}, cloud_first_mode={}, retry_max={}, " +
135- "retry_backoff_ms={}, retry_max_backoff_ms={}" ,
136- graphName , this .storageClient .provider (), this .bucket , this .objectPrefix ,
137- this .syncIntervalSeconds , this .syncIncremental , this .cloudFirstMode ,
138- this .syncRetryMax , this .syncRetryBackoffMs , this .syncRetryMaxBackoffMs );
141+ startSstWatchSync ();
142+ startPeriodicSync ();
143+ log .info ("RocksDB cloud enabled for graph {}: {}://{}/{}, interval={}s, " +
144+ "incremental={}, synchronous_sst_upload_mode={}, " +
145+ "retry_max={}, retry_backoff_ms={}, retry_max_backoff_ms={}" ,
146+ graphName , this .storageClient .provider (), this .bucket , this .objectPrefix ,
147+ this .syncIntervalSeconds , this .syncIncremental ,
148+ this .synchronousSstUploadMode , this .syncRetryMax , this .syncRetryBackoffMs ,
149+ this .syncRetryMaxBackoffMs );
139150 }
140151
141152 @ Override
@@ -232,6 +243,7 @@ private static boolean nonRecoverableReadError(Throwable t) {
232243
233244 @ Override
234245 void shutdown () {
246+ stopSstWatchSync ();
235247 stopPeriodicSync ();
236248 try {
237249 syncNow (true , true );
@@ -262,6 +274,101 @@ private void startPeriodicSync() {
262274 }, this .syncIntervalSeconds , this .syncIntervalSeconds , TimeUnit .SECONDS );
263275 }
264276
277+ private void startSstWatchSync () {
278+ // Single-flag behavior: only synchronous_sst_upload_mode=true enables SST-triggered uploads.
279+ if (!this .synchronousSstUploadMode ) {
280+ return ;
281+ }
282+ try {
283+ this .sstWatchService = FileSystems .getDefault ().newWatchService ();
284+ Path dbPath = Paths .get (getDbPath ());
285+ dbPath .register (this .sstWatchService ,
286+ StandardWatchEventKinds .ENTRY_CREATE ,
287+ StandardWatchEventKinds .ENTRY_MODIFY ,
288+ StandardWatchEventKinds .ENTRY_DELETE );
289+ } catch (Exception e ) {
290+ log .warn ("Failed to start SST watch sync for {}: {}" ,
291+ getGraphName (), e .getMessage ());
292+ return ;
293+ }
294+
295+ this .sstWatchThread = new Thread (() -> {
296+ while (!Thread .currentThread ().isInterrupted ()) {
297+ try {
298+ WatchKey key = this .sstWatchService .poll (1 , TimeUnit .SECONDS );
299+ if (key == null ) {
300+ continue ;
301+ }
302+
303+ boolean hasSstChange = false ;
304+ for (WatchEvent <?> event : key .pollEvents ()) {
305+ if (event .kind () == StandardWatchEventKinds .OVERFLOW ) {
306+ hasSstChange = true ;
307+ continue ;
308+ }
309+ Object context = event .context ();
310+ if (!(context instanceof Path )) {
311+ continue ;
312+ }
313+ String fileName = ((Path ) context ).getFileName ().toString ()
314+ .toLowerCase (Locale .ROOT );
315+ if (fileName .endsWith (".sst" )) {
316+ hasSstChange = true ;
317+ break ;
318+ }
319+ }
320+ if (!key .reset ()) {
321+ break ;
322+ }
323+
324+ if (hasSstChange ) {
325+ queueSstSync ();
326+ }
327+ } catch (InterruptedException e ) {
328+ Thread .currentThread ().interrupt ();
329+ break ;
330+ } catch (Throwable t ) {
331+ log .warn ("SST watch sync loop failed for {}: {}" ,
332+ getGraphName (), t .getMessage ());
333+ }
334+ }
335+ }, "store-rocksdb-sst-watch-" + getGraphName ());
336+ this .sstWatchThread .setDaemon (true );
337+ this .sstWatchThread .start ();
338+ }
339+
340+ private void queueSstSync () {
341+ if (!this .sstSyncQueued .compareAndSet (false , true )) {
342+ return ;
343+ }
344+
345+ // Synchronous-only path: upload SST-triggered changes immediately.
346+ try {
347+ syncNow (false , false );
348+ log .debug ("Synchronous SST cloud upload completed for graph {}" , getGraphName ());
349+ } catch (Throwable t ) {
350+ log .warn ("Synchronous SST cloud upload failed for {}: {}" ,
351+ getGraphName (), t .getMessage ());
352+ } finally {
353+ this .sstSyncQueued .set (false );
354+ }
355+ }
356+
357+ private void stopSstWatchSync () {
358+ if (this .sstWatchThread != null ) {
359+ this .sstWatchThread .interrupt ();
360+ this .sstWatchThread = null ;
361+ }
362+ if (this .sstWatchService != null ) {
363+ try {
364+ this .sstWatchService .close ();
365+ } catch (Exception ignored ) {
366+ // Ignore close exception on shutdown path
367+ }
368+ this .sstWatchService = null ;
369+ }
370+ }
371+
265372 private void stopPeriodicSync () {
266373 if (this .periodicSyncFuture != null && !this .periodicSyncFuture .isCancelled ()) {
267374 this .periodicSyncFuture .cancel (false );
@@ -306,8 +413,8 @@ private static String getString(HugeConfig conf, String defaultValue,
306413 }
307414
308415 private static boolean getBoolean (HugeConfig conf , String key ,
309- String legacyKey , boolean defaultValue ) {
310- return Boolean .parseBoolean (getString (conf , String .valueOf (defaultValue ), key , legacyKey ));
416+ String legacyKey ) {
417+ return Boolean .parseBoolean (getString (conf , String .valueOf (true ), key , legacyKey ));
311418 }
312419
313420 private static int getInt (HugeConfig conf , String key ,
@@ -346,14 +453,7 @@ private <T> T withReadHydrationRetry(Op<T> primary, Op<T> retry) throws DBStoreE
346453
347454 @ Override
348455 public Integer commit () throws DBStoreException {
349- Integer count = super .commit ();
350- if (count != null && count > 0 ) {
351- if (this .cloudSession .cloudFirstMode ) {
352- // In cloud-first mode, sync before acknowledging commit to caller.
353- this .cloudSession .syncNow (false , true );
354- }
355- }
356- return count ;
456+ return super .commit ();
357457 }
358458
359459 @ Override
0 commit comments