main, fuseadapter: stop gracefully on SIGTERM/SIGINT and detach stale FUSE mounts

README.md

@@ -220,7 +220,7 @@ printf 'hello\n' > /tmp/elastickv-mount/docs/hello.txt
 cat /tmp/elastickv-mount/docs/hello.txt
 ```
 
-The root owner and mode flags apply when the filesystem root is first created; the defaults are UID/GID `0` and mode `0755`, and the mount enforces `default_permissions`, so set them to the user that will write to the mount. `--filesystemClientID` defaults to `--raftId` and identifies open-handle leases. `--filesystemCapacity` and `--filesystemMaxFiles` set values reported by `statfs`; they do not enforce write quotas. The server mounts through `fusermount3` (package `fuse3`) after its network listeners are up. Stopping the process with a signal leaves the mount attached in a `Transport endpoint is not connected` state; run `fusermount3 -uz <mount-point>` before restarting.
+The root owner and mode flags apply when the filesystem root is first created; the defaults are UID/GID `0` and mode `0755`, and the mount enforces `default_permissions`, so set them to the user that will write to the mount. `--filesystemClientID` defaults to `--raftId` and identifies open-handle leases. `--filesystemCapacity` and `--filesystemMaxFiles` set values reported by `statfs`; they do not enforce write quotas. The server mounts through `fusermount3` (package `fuse3`) after its network listeners are up. The server unmounts the filesystem when it stops, including on `SIGTERM`/`SIGINT`, and a restart after `SIGKILL` or a crash detaches the dead mount it finds at the mount point before mounting again. See the [FUSE filesystem guide](docs/filesystem.md) for details.
 
 This FUSE implementation supports regular files and directories, but not cross-directory rename, hard links, symbolic links, extended attributes, or full POSIX locking. Concurrent modifications from different mounts are resolved optimistically: the loser gets `EAGAIN`. Most operations can simply be retried, but `close` must not be, and an `O_APPEND` write that other mounts may also be appending to must not be either; the guide's Concurrency section has the per-operation rules. See the [FUSE filesystem guide](docs/filesystem.md) for prerequisites, multi-node setup, semantics, monitoring, and troubleshooting, and the [filesystem design](docs/design/2026_02_24_implemented_filesystem_on_elastickv.md) for the supported operations and placement model.
 

adapter/background_reservation_shutdown_test.go

@@ -0,0 +1,42 @@
+package adapter
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBackgroundReservationStartsLiveAfterShutdownBound(t *testing.T) {
+	w := newBackgroundWorkers()
+	r, ok := w.reserve()
+	require.True(t, ok)
+	done := make(chan struct{})
+	go func() { w.stop("test", time.Millisecond, time.Second); close(done) }()
+	<-w.ctx.Done()
+	ran := make(chan error, 1)
+	r.start(time.Second, func(ctx context.Context) { ran <- ctx.Err() })
+	require.NoError(t, <-ran)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("stop did not join reserved cleanup")
+	}
+}
+
+func TestBackgroundReservationCannotStartAfterStopReturns(t *testing.T) {
+	w := newBackgroundWorkers()
+	r, ok := w.reserve()
+	require.True(t, ok)
+	w.stop("test", time.Millisecond, time.Millisecond)
+	ran := make(chan struct{}, 1)
+	r.start(time.Second, func(context.Context) { ran <- struct{}{} })
+	r.release()
+	require.Zero(t, w.gate.remaining())
+	select {
+	case <-ran:
+		t.Fatal("cleanup started after shutdown returned")
+	case <-time.After(time.Millisecond):
+	}
+}

adapter/background_workers.go

@@ -0,0 +1,150 @@
+package adapter
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+)
+
+// backgroundWorkers runs the jobs a request starts but does not wait for,
+// such as the cleanup of a deleted table's old generation. Each job is
+// counted in an inflightGate so the server's Stop can join it, bounded,
+// before the stores close underneath it. Stop cancels jobs already started
+// once the bound has passed; reservations that commit later get a live
+// timeout and stay counted in the final join. Reservations cannot start
+// cleanup after that final join ends. This bounded drain does not provide
+// durable recovery of cleanup for mutations that outlive shutdown.
+// A nil *backgroundWorkers
+// runs every job detached under context.Background() and has nothing to
+// stop, which keeps zero-value servers in tests working.
+type backgroundWorkers struct {
+	gate    *inflightGate
+	ctx     context.Context
+	cancel  context.CancelFunc
+	mu      sync.Mutex
+	running map[*backgroundReservation]context.CancelFunc
+	closed  bool
+}
+
+func newBackgroundWorkers() *backgroundWorkers {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &backgroundWorkers{gate: newInflightGate(), ctx: ctx, cancel: cancel, running: make(map[*backgroundReservation]context.CancelFunc)}
+}
+
+// backgroundReservation is a job slot claimed before the commit that makes
+// the job necessary, so the job can still start after that commit even when
+// stop began in between: stop waits for the slot as for a running job. The
+// holder either starts the job through it or releases it.
+type backgroundReservation struct {
+	mu   sync.Mutex
+	w    *backgroundWorkers
+	used bool
+}
+
+// reserve claims a job slot, or reports false once stop has begun, in which
+// case the caller must not commit the change the job would follow up on.
+func (w *backgroundWorkers) reserve() (*backgroundReservation, bool) {
+	if w != nil && !w.gate.enter() {
+		return nil, false
+	}
+	return &backgroundReservation{w: w}, true
+}
+
+// start runs run on its own goroutine under a context bounded by timeout,
+// handing it the reserved slot. A reservation starts at most one job.
+func (r *backgroundReservation) start(timeout time.Duration, run func(ctx context.Context)) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.used {
+		return
+	}
+	r.used = true
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	if r.w != nil {
+		r.w.mu.Lock()
+		if r.w.closed {
+			r.w.mu.Unlock()
+			cancel()
+			r.w.gate.exit()
+			return
+		}
+		r.w.running[r] = cancel
+		r.w.mu.Unlock()
+	}
+	go func() {
+		defer cancel()
+		if r.w != nil {
+			defer r.w.gate.exit()
+			defer func() {
+				r.w.mu.Lock()
+				delete(r.w.running, r)
+				r.w.mu.Unlock()
+			}()
+		}
+		run(ctx)
+	}()
+}
+
+// release gives back a slot no job was started through; it is a no-op once
+// start ran or after an earlier release.
+func (r *backgroundReservation) release() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.used {
+		return
+	}
+	r.used = true
+	if r.w != nil {
+		r.w.gate.exit()
+	}
+}
+
+// stop refuses new reservations, waits up to timeout for the reserved slots
+// and the running jobs, then cancels the context of any jobs that remain and
+// waits up to joinTimeout for them to return, so none is still dispatching
+// when the stores close. A call made while an earlier one is still waiting
+// waits until the same deadlines (see inflightGate); name labels the
+// warnings it logs.
+func (w *backgroundWorkers) stop(name string, timeout, joinTimeout time.Duration) {
+	if w == nil {
+		return
+	}
+	defer w.closeReservations()
+	finished := w.gate.drain(timeout)
+	w.cancelRunning()
+	if finished {
+		return
+	}
+	slog.Warn(name+": background jobs still running after the shutdown bound, cancelling them",
+		slog.Int64("in_flight", w.gate.remaining()),
+		slog.Duration("timeout", timeout),
+	)
+	if !w.gate.join(joinTimeout) {
+		slog.Warn(name+": cancelled background jobs did not return, continuing shutdown",
+			slog.Int64("in_flight", w.gate.remaining()),
+			slog.Duration("timeout", joinTimeout),
+		)
+	}
+}
+
+func (w *backgroundWorkers) closeReservations() {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.closed = true
+	for _, cancel := range w.running {
+		cancel()
+	}
+}
+
+func (w *backgroundWorkers) cancelRunning() {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.cancel()
+	// Reservations still owned by a mutation are not jobs yet. If that
+	// mutation commits later, its cleanup gets a live, independently bounded
+	// context and remains counted in the join window.
+	for _, cancel := range w.running {
+		cancel()
+	}
+}

adapter/dynamodb.go

@@ -122,6 +122,7 @@ type DynamoDBServer struct {
 	store           store.MVCCStore
 	coordinator     kv.Coordinator
 	httpServer      *http.Server
+	httpConns       *httpConnTracker
 	targetHandlers  map[string]func(http.ResponseWriter, *http.Request)
 	readTracker     *kv.ActiveTimestampTracker
 	requestObserver monitoring.DynamoDBRequestObserver
@@ -138,6 +139,13 @@ type DynamoDBServer struct {
 	// Set ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=0 or pass
 	// WithDynamoOnePhaseTxnDedup(false) for an operator rollback.
 	onePhaseTxnDedup bool
+
+	// tableCleanups runs the cleanup of a deleted table's old generation
+	// (see launchDeletedTableCleanup) so Stop can join it, bounded, before
+	// the stores close underneath it. Its slot is reserved before the change
+	// that orphans the generation commits, whichever surface (public
+	// listener, admin HTTP or gRPC forwarder) the change came through.
+	tableCleanups *backgroundWorkers
 }
 
 // WithDynamoDBRequestObserver enables Prometheus-compatible request metrics.
@@ -264,6 +272,7 @@ func NewDynamoDBServer(listen net.Listener, st store.MVCCStore, coordinate kv.Co
 		store:            st,
 		coordinator:      kv.WithKeyVizLabel(coordinate, keyviz.LabelDynamo),
 		onePhaseTxnDedup: os.Getenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP") != "0",
+		tableCleanups:    newBackgroundWorkers(),
 	}
 	for i := range d.itemUpdateLocks {
 		d.itemUpdateLocks[i] = make(chan struct{}, 1)
@@ -287,6 +296,7 @@ func NewDynamoDBServer(listen net.Listener, st store.MVCCStore, coordinate kv.Co
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", d.handle)
 	d.httpServer = &http.Server{Handler: mux, ReadHeaderTimeout: time.Second}
+	d.httpConns = trackHTTPConns(d.httpServer)
 	for _, opt := range opts {
 		if opt != nil {
 			opt(d)
@@ -308,12 +318,36 @@ func (d *DynamoDBServer) Run() error {
 	return nil
 }
 
+// Stop drains the HTTP listener (bounded, see shutdownHTTPServer), then
+// waits, under one more bound, for the deleted-table cleanups already
+// started or reserved by a delete still committing, cancelling and briefly
+// joining any still running, so none is dispatching when the stores close
+// after Stop. From then on a change that would need a cleanup is refused
+// with ErrDynamoServerStopping.
 func (d *DynamoDBServer) Stop() {
 	if d.httpServer != nil {
-		_ = d.httpServer.Shutdown(context.Background())
+		shutdownHTTPServer(d.httpServer, d.httpConns, "dynamodb")
 	}
+	d.tableCleanups.stop("dynamodb server", httpShutdownTimeout, dynamoTableCleanupJoinTimeout)
 }
 
+// ErrDynamoServerStopping is returned by a table delete, or a legacy-table
+// migration step, that would orphan a table generation once Stop has begun:
+// the generation's cleanup can no longer start, so the change is refused
+// before it commits, and the table is left as it was. The public listener
+// answers it with 503 ServiceUnavailable, which clients retry. It is also
+// marked as ErrAdminNotLeader, so the admin bridge answers it with the
+// not-leader response and the admin client retries against another node.
+var ErrDynamoServerStopping = errors.Mark(
+	newDynamoAPIError(http.StatusServiceUnavailable, dynamoErrServiceUnavailable, "server is stopping"),
+	ErrAdminNotLeader,
+)
+
+// dynamoTableCleanupJoinTimeout bounds how long Stop waits, after
+// cancelling the table cleanups that outlived the shutdown bound, for them
+// to return. A variable so tests can shorten it.
+var dynamoTableCleanupJoinTimeout = 2 * time.Second
+
 // proxyToLeader forwards the HTTP request to the current DynamoDB leader when
 // this node is not the Raft leader.  Returns true if the request was handled
 // (either proxied or an error response was written), false if the request

adapter/dynamodb_migration.go

@@ -318,6 +318,13 @@ func (d *DynamoDBServer) finalizeLegacyTableMigration(ctx context.Context, schem
 	if err != nil {
 		return errors.WithStack(err)
 	}
+	// As for a table delete, the old generation's cleanup slot is claimed
+	// before the schema switch can commit.
+	cleanup, ok := d.tableCleanups.reserve()
+	if !ok {
+		return errors.WithStack(ErrDynamoServerStopping)
+	}
+	defer cleanup.release()
 	readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb finalize migration: begin read timestamp", dynamoTableMetaKey(schema.TableName))
 	if err != nil {
 		return errors.WithStack(err)
@@ -334,6 +341,6 @@ func (d *DynamoDBServer) finalizeLegacyTableMigration(ctx context.Context, schem
 	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
 		return errors.WithStack(err)
 	}
-	d.launchDeletedTableCleanup(schema.TableName, oldGeneration)
+	d.launchDeletedTableCleanup(cleanup, schema.TableName, oldGeneration)
 	return nil
 }

adapter/dynamodb_schema.go

@@ -295,6 +295,14 @@ func decodeDeleteTableInput(bodyReader io.Reader) (deleteTableInput, error) {
 }
 
 func (d *DynamoDBServer) deleteTableWithRetry(ctx context.Context, tableName string) error {
+	// Claim the cleanup's slot before the metadata delete can commit: once
+	// Stop has begun the cleanup could no longer start, and a committed
+	// delete without it would leave the generation's items behind.
+	cleanup, ok := d.tableCleanups.reserve()
+	if !ok {
+		return errors.WithStack(ErrDynamoServerStopping)
+	}
+	defer cleanup.release()
 	backoff := transactRetryInitialBackoff
 	deadline := time.Now().Add(transactRetryMaxDuration)
 	for range transactRetryMaxAttempts {
@@ -324,7 +332,7 @@ func (d *DynamoDBServer) deleteTableWithRetry(ctx context.Context, tableName str
 				return errors.WithStack(err)
 			}
 		} else {
-			d.launchDeletedTableCleanup(tableName, schema.Generation)
+			d.launchDeletedTableCleanup(cleanup, tableName, schema.Generation)
 			return nil
 		}
 
@@ -336,18 +344,19 @@ func (d *DynamoDBServer) deleteTableWithRetry(ctx context.Context, tableName str
 	return newDynamoAPIError(http.StatusInternalServerError, dynamoErrInternal, "delete table retry attempts exhausted")
 }
 
-func (d *DynamoDBServer) launchDeletedTableCleanup(tableName string, generation uint64) {
-	go func() {
-		ctx, cancel := context.WithTimeout(context.Background(), tableCleanupAsyncTimeout)
-		defer cancel()
+// launchDeletedTableCleanup removes a deleted generation's items, seals and
+// index entries in the background, through the slot the caller reserved
+// before committing the change that orphaned them.
+func (d *DynamoDBServer) launchDeletedTableCleanup(cleanup *backgroundReservation, tableName string, generation uint64) {
+	cleanup.start(tableCleanupAsyncTimeout, func(ctx context.Context) {
 		if err := d.cleanupDeletedTableGeneration(ctx, tableName, generation); err != nil {
 			slog.Error("dynamodb delete table cleanup failed",
 				"table", tableName,
 				"generation", generation,
 				"error", err,
 			)
 		}
-	}()
+	})
 }
 
 func (d *DynamoDBServer) cleanupDeletedTableGeneration(ctx context.Context, tableName string, generation uint64) error {