Method CreateSession

duckdbservice/service.go:962–1195

CreateSession creates a new DuckDB session for the given username. secretStatements are the user's persistent secrets to replay (shared-warm mode only); replay failures come back as warnings — logged on the worker and again by the control plane's session manager, never failing the session — rather t

(username, memoryLimit string, threads int, secretStatements []string)

		}

		slog.Info("Starting DuckDB service", "network", network, "addr", addr)
	}

	// Handle graceful shutdown
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		<-sigChan
		slog.Info("Draining DuckDB service before shutdown...")
		svc.BeginDrain()
		ctx, cancel := context.WithTimeout(context.Background(), workerShutdownDrainTime)
		if !svc.WaitForDrain(ctx) {
			slog.Warn("DuckDB service drain timed out before shutdown.", "timeout", workerShutdownDrainTime)
			cancel()
			// Close first so teardown WARNs land in the batch, then flush.
			svc.CloseAll()
			cliboot.FlushLogging()
			os.Exit(0)
		}
		cancel()
		slog.Info("Shutting down DuckDB service...")
		svc.Shutdown()
		cliboot.FlushLogging()
		os.Exit(0)
	}()

	if err := svc.Serve(listener); err != nil {
		slog.Error("DuckDB service error", "error", err)
		os.Exit(1)
	}
}

// Serve starts serving on the given listener.
func (svc *DuckDBService) Serve(listener net.Listener) error {
	handler := NewFlightSQLHandler(svc.pool)

	opts, err := flightServerOptions(svc.cfg, listener)
	if err != nil {
		return err
	}

	// Wrap the flightsql server with custom action handling.
	// flightsql.NewFlightServer routes standard Flight SQL actions but rejects
	// custom action types. Our wrapper intercepts custom actions (CreateSession,
	// DestroySession, HealthCheck) before falling through to the standard router.
	flightSqlSrv := flightsql.NewFlightServer(handler)
	customSrv := &customActionServer{FlightServer: flightSqlSrv, handler: handler}

	svc.flightSrv = flight.NewServerWithMiddleware(nil, opts...)
	svc.flightSrv.RegisterFlightService(customSrv)
	svc.flightSrv.InitListener(listener)
	return svc.flightSrv.Serve()
}

func flightServerOptions(cfg ServiceConfig, listener net.Listener) ([]grpc.ServerOption, error) {
	opts := []grpc.ServerOption{
		grpc.MaxRecvMsgSize(flightclient.MaxGRPCMessageSize),