Add server-managed feature SQLite storage
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -10,6 +10,7 @@ pub mod backend_job;
|
|||||||
pub mod companion;
|
pub mod companion;
|
||||||
pub mod config;
|
pub mod config;
|
||||||
pub mod config_source;
|
pub mod config_source;
|
||||||
|
pub mod feature_storage;
|
||||||
pub mod hosts;
|
pub mod hosts;
|
||||||
pub mod memory_backend;
|
pub mod memory_backend;
|
||||||
pub mod memory_staging;
|
pub mod memory_staging;
|
||||||
@@ -43,6 +44,10 @@ pub use authority::{
|
|||||||
WorkspaceAuthority,
|
WorkspaceAuthority,
|
||||||
};
|
};
|
||||||
pub use config::{ResolvedWorkspaceBackendConfig, ServerHostConfigFile};
|
pub use config::{ResolvedWorkspaceBackendConfig, ServerHostConfigFile};
|
||||||
|
pub use feature_storage::{
|
||||||
|
FeatureDatabase, FeatureMigration, FeatureRegistration, FeatureStorage, FeatureStorageError,
|
||||||
|
RegisteredFeature, WorkspaceFeatureStorage,
|
||||||
|
};
|
||||||
pub use records::{ObjectiveDetail, ObjectiveSummary, TicketDetail, TicketSummary};
|
pub use records::{ObjectiveDetail, ObjectiveSummary, TicketDetail, TicketSummary};
|
||||||
pub use repositories::{ConfiguredRepository, RepositoryLogRead, RepositoryRegistryReader};
|
pub use repositories::{ConfiguredRepository, RepositoryLogRead, RepositoryRegistryReader};
|
||||||
pub use server::{
|
pub use server::{
|
||||||
|
|||||||
@@ -1387,6 +1387,7 @@ pub struct WorkspaceApi {
|
|||||||
signing_identities: WorkspaceSigningIdentityService,
|
signing_identities: WorkspaceSigningIdentityService,
|
||||||
config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry,
|
config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry,
|
||||||
prompt_projection_cache: crate::prompt_settings::WorkspacePromptProjectionCache,
|
prompt_projection_cache: crate::prompt_settings::WorkspacePromptProjectionCache,
|
||||||
|
feature_storage: crate::WorkspaceFeatureStorage,
|
||||||
authority: SqliteWorkspaceAuthority,
|
authority: SqliteWorkspaceAuthority,
|
||||||
_worker_projection_shutdown: Arc<WorkerProjectionShutdownGuard>,
|
_worker_projection_shutdown: Arc<WorkerProjectionShutdownGuard>,
|
||||||
runtime: Arc<RuntimeRegistry>,
|
runtime: Arc<RuntimeRegistry>,
|
||||||
@@ -2257,6 +2258,11 @@ impl WorkspaceServerApi {
|
|||||||
None,
|
None,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
api.feature_storage.delete().map_err(|error| {
|
||||||
|
Error::Store(format!(
|
||||||
|
"failed to delete Workspace Feature storage: {error}"
|
||||||
|
))
|
||||||
|
})?;
|
||||||
WorkspaceSigningIdentityService::new(self.store.clone(), self.signing_materials.clone())
|
WorkspaceSigningIdentityService::new(self.store.clone(), self.signing_materials.clone())
|
||||||
.delete_material(&operation.workspace_id)?;
|
.delete_material(&operation.workspace_id)?;
|
||||||
let completed = self.store.finalize_workspace_deletion(operation_id)?;
|
let completed = self.store.finalize_workspace_deletion(operation_id)?;
|
||||||
@@ -2303,6 +2309,16 @@ impl WorkspaceServerApi {
|
|||||||
Ok(Some(router))
|
Ok(Some(router))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn shutdown_feature_storage(&self) -> Result<()> {
|
||||||
|
let apis = self.apis.lock().await.values().cloned().collect::<Vec<_>>();
|
||||||
|
for api in apis {
|
||||||
|
api.feature_storage.shutdown().map_err(|error| {
|
||||||
|
Error::Store(format!("failed to shut down Feature storage: {error}"))
|
||||||
|
})?;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
async fn preload(&self) -> Result<()> {
|
async fn preload(&self) -> Result<()> {
|
||||||
for workspace in self.store.list_workspaces()? {
|
for workspace in self.store.list_workspaces()? {
|
||||||
let _ = self
|
let _ = self
|
||||||
@@ -2836,6 +2852,10 @@ pub async fn build_workspace_server_router(
|
|||||||
let api = WorkspaceServerApi::new(template, store);
|
let api = WorkspaceServerApi::new(template, store);
|
||||||
api.preload().await?;
|
api.preload().await?;
|
||||||
api.recover_workspace_deletions().await?;
|
api.recover_workspace_deletions().await?;
|
||||||
|
Ok(workspace_server_router(api))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn workspace_server_router(api: WorkspaceServerApi) -> Router {
|
||||||
let contract_service = ServerApiContractService::Server(api.clone());
|
let contract_service = ServerApiContractService::Server(api.clone());
|
||||||
let catalog = Router::new()
|
let catalog = Router::new()
|
||||||
.fallback(dispatch_workspace_request)
|
.fallback(dispatch_workspace_request)
|
||||||
@@ -2847,10 +2867,10 @@ pub async fn build_workspace_server_router(
|
|||||||
.merge(generated_workspace_catalog_contract_router(
|
.merge(generated_workspace_catalog_contract_router(
|
||||||
contract_service,
|
contract_service,
|
||||||
));
|
));
|
||||||
Ok(catalog.layer(axum::middleware::from_fn_with_state(
|
catalog.layer(axum::middleware::from_fn_with_state(
|
||||||
api,
|
api,
|
||||||
enforce_server_cookie_mutation_origin,
|
enforce_server_cookie_mutation_origin,
|
||||||
)))
|
))
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
@@ -2928,6 +2948,12 @@ fn embedded_runtime_request_audience(config: &ServerConfig) -> crate::Result<Str
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl WorkspaceApi {
|
impl WorkspaceApi {
|
||||||
|
/// Server-managed, Workspace-scoped storage for trusted Feature repositories.
|
||||||
|
/// Feature registration and SQL access stay inside the Server process.
|
||||||
|
pub fn feature_storage(&self) -> &crate::WorkspaceFeatureStorage {
|
||||||
|
&self.feature_storage
|
||||||
|
}
|
||||||
|
|
||||||
pub fn with_config_schema_provider(
|
pub fn with_config_schema_provider(
|
||||||
mut self,
|
mut self,
|
||||||
provider: Arc<dyn crate::config_source::WorkspaceConfigSchemaProvider>,
|
provider: Arc<dyn crate::config_source::WorkspaceConfigSchemaProvider>,
|
||||||
@@ -3184,6 +3210,11 @@ impl WorkspaceApi {
|
|||||||
.await
|
.await
|
||||||
.map_err(|error| Error::InvalidInput(error.to_string()))?;
|
.map_err(|error| Error::InvalidInput(error.to_string()))?;
|
||||||
}
|
}
|
||||||
|
let feature_storage = crate::FeatureStorage::for_server_database(&config.database_path)
|
||||||
|
.and_then(|storage| storage.workspace(&config.workspace_id))
|
||||||
|
.map_err(|error| {
|
||||||
|
Error::Store(format!("failed to initialize Feature storage: {error}"))
|
||||||
|
})?;
|
||||||
let api = Self {
|
let api = Self {
|
||||||
config_store,
|
config_store,
|
||||||
repository_secrets,
|
repository_secrets,
|
||||||
@@ -3191,6 +3222,7 @@ impl WorkspaceApi {
|
|||||||
config_schema_registry,
|
config_schema_registry,
|
||||||
prompt_projection_cache:
|
prompt_projection_cache:
|
||||||
crate::prompt_settings::WorkspacePromptProjectionCache::default(),
|
crate::prompt_settings::WorkspacePromptProjectionCache::default(),
|
||||||
|
feature_storage,
|
||||||
authority: SqliteWorkspaceAuthority::new(
|
authority: SqliteWorkspaceAuthority::new(
|
||||||
config.database_path.clone(),
|
config.database_path.clone(),
|
||||||
config.workspace_id.clone(),
|
config.workspace_id.clone(),
|
||||||
@@ -10025,10 +10057,15 @@ pub async fn serve_workspace_catalog_with_shutdown<F>(
|
|||||||
where
|
where
|
||||||
F: std::future::Future<Output = ()> + Send + 'static,
|
F: std::future::Future<Output = ()> + Send + 'static,
|
||||||
{
|
{
|
||||||
let router = build_workspace_server_router(template, store).await?;
|
let api = WorkspaceServerApi::new(template, store);
|
||||||
axum::serve(listener, router)
|
api.preload().await?;
|
||||||
|
api.recover_workspace_deletions().await?;
|
||||||
|
let result = axum::serve(listener, workspace_server_router(api.clone()))
|
||||||
.with_graceful_shutdown(shutdown)
|
.with_graceful_shutdown(shutdown)
|
||||||
.await?;
|
.await;
|
||||||
|
let storage_result = api.shutdown_feature_storage().await;
|
||||||
|
result?;
|
||||||
|
storage_result?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -10037,8 +10074,13 @@ pub async fn serve_workspace_catalog(
|
|||||||
store: Arc<dyn ControlPlaneStore>,
|
store: Arc<dyn ControlPlaneStore>,
|
||||||
listener: TcpListener,
|
listener: TcpListener,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let router = build_workspace_server_router(template, store).await?;
|
let api = WorkspaceServerApi::new(template, store);
|
||||||
axum::serve(listener, router).await?;
|
api.preload().await?;
|
||||||
|
api.recover_workspace_deletions().await?;
|
||||||
|
let result = axum::serve(listener, workspace_server_router(api.clone())).await;
|
||||||
|
let storage_result = api.shutdown_feature_storage().await;
|
||||||
|
result?;
|
||||||
|
storage_result?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -10056,9 +10098,14 @@ pub async fn serve(
|
|||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let api = WorkspaceApi::new(config, store).await?;
|
let api = WorkspaceApi::new(config, store).await?;
|
||||||
let orchestrator_hook = tokio::spawn(run_orchestrator_turn_end_hook(api.clone()));
|
let orchestrator_hook = tokio::spawn(run_orchestrator_turn_end_hook(api.clone()));
|
||||||
let result = axum::serve(listener, build_router(api)).await;
|
let result = axum::serve(listener, build_router(api.clone())).await;
|
||||||
orchestrator_hook.abort();
|
orchestrator_hook.abort();
|
||||||
|
let storage_result = api
|
||||||
|
.feature_storage
|
||||||
|
.shutdown()
|
||||||
|
.map_err(|error| Error::Store(format!("failed to shut down Feature storage: {error}")));
|
||||||
result?;
|
result?;
|
||||||
|
storage_result?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -29,6 +29,7 @@ It is not a dumping ground for external research, old plans, API inventories, or
|
|||||||
21. [`development/validation.md`](development/validation.md) — how to check changes.
|
21. [`development/validation.md`](development/validation.md) — how to check changes.
|
||||||
22. [`development/workspace-schema-migrations.md`](development/workspace-schema-migrations.md) — how to preflight, apply, verify, and roll back control-plane SQLite schema changes.
|
22. [`development/workspace-schema-migrations.md`](development/workspace-schema-migrations.md) — how to preflight, apply, verify, and roll back control-plane SQLite schema changes.
|
||||||
23. [`design/standalone-agent-host.md`](design/standalone-agent-host.md) — in-process standalone Worker host の依存方向、authority、lifecycle、非目標。
|
23. [`design/standalone-agent-host.md`](design/standalone-agent-host.md) — in-process standalone Worker host の依存方向、authority、lifecycle、非目標。
|
||||||
|
24. [`design/server-feature-storage.md`](design/server-feature-storage.md) — Server管理のWorkspace/Feature専用SQLite、migration、transaction、backup/restore、削除・shutdown境界。
|
||||||
|
|
||||||
## What belongs here
|
## What belongs here
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,143 @@
|
|||||||
|
# Server-managed Feature storage
|
||||||
|
|
||||||
|
`yoi-workspace-server::feature_storage` provides the SQLite lifecycle boundary for trusted Server-side Features. It is not a Worker feature API and it is not a model-visible SQL tool.
|
||||||
|
|
||||||
|
## Ownership boundary
|
||||||
|
|
||||||
|
The Server owns:
|
||||||
|
|
||||||
|
- the physical layout under `feature-storage/<workspace-id>/features/<feature-id>.sqlite`;
|
||||||
|
- validated Workspace and Feature identifiers (no caller-provided database path);
|
||||||
|
- connection reuse, `foreign_keys`, WAL, `busy_timeout`, and write serialization;
|
||||||
|
- migration timing/history, shutdown admission, online backup/restore, and Workspace deletion fencing.
|
||||||
|
|
||||||
|
A Feature owns:
|
||||||
|
|
||||||
|
- its tables, migration bodies, queries, and domain transactions;
|
||||||
|
- validation and typed repository/API methods exposed to its Server handler;
|
||||||
|
- authorization before the repository is called.
|
||||||
|
|
||||||
|
A Feature database is an API/ownership isolation boundary inside the trusted Server process. It does not sandbox arbitrary native code and does not replace Workspace authorization. A `FeatureDatabase`, SQLite path, or general SQL endpoint must never be sent to a Worker/Runtime or added to a model tool contract.
|
||||||
|
|
||||||
|
## Registration and repository example
|
||||||
|
|
||||||
|
Registration is trusted startup code. Merely constructing or registering storage is lazy and creates no directory or database; the first `open` applies migrations.
|
||||||
|
|
||||||
|
```rust
|
||||||
|
use rusqlite::{Transaction, params};
|
||||||
|
use yoi_workspace_server::{
|
||||||
|
FeatureDatabase, FeatureMigration, FeatureRegistration, FeatureStorageError,
|
||||||
|
WorkspaceApi,
|
||||||
|
};
|
||||||
|
|
||||||
|
type StorageResult<T> =
|
||||||
|
std::result::Result<T, FeatureStorageError>;
|
||||||
|
|
||||||
|
fn create_subject_memory(tx: &Transaction<'_>) -> StorageResult<()> {
|
||||||
|
tx.execute_batch(
|
||||||
|
"CREATE TABLE subjects (
|
||||||
|
subject_id TEXT PRIMARY KEY,
|
||||||
|
revision INTEGER NOT NULL
|
||||||
|
);
|
||||||
|
CREATE TABLE memories (
|
||||||
|
subject_id TEXT NOT NULL,
|
||||||
|
memory_id TEXT NOT NULL,
|
||||||
|
body TEXT NOT NULL,
|
||||||
|
PRIMARY KEY (subject_id, memory_id),
|
||||||
|
FOREIGN KEY (subject_id) REFERENCES subjects(subject_id)
|
||||||
|
);
|
||||||
|
CREATE TABLE interventions (
|
||||||
|
intervention_id TEXT PRIMARY KEY,
|
||||||
|
subject_id TEXT NOT NULL,
|
||||||
|
detail TEXT NOT NULL,
|
||||||
|
FOREIGN KEY (subject_id) REFERENCES subjects(subject_id)
|
||||||
|
);"
|
||||||
|
)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
static MIGRATIONS: &[FeatureMigration] = &[
|
||||||
|
FeatureMigration::new(1, "subject memory baseline", create_subject_memory),
|
||||||
|
];
|
||||||
|
|
||||||
|
const REGISTRATION: FeatureRegistration =
|
||||||
|
FeatureRegistration::new("subjektiv", MIGRATIONS);
|
||||||
|
|
||||||
|
struct SubjektivRepository {
|
||||||
|
database: FeatureDatabase,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SubjektivRepository {
|
||||||
|
fn apply_candidate(
|
||||||
|
&self,
|
||||||
|
subject_id: &str,
|
||||||
|
memory_id: &str,
|
||||||
|
body: &str,
|
||||||
|
intervention_id: &str,
|
||||||
|
) -> StorageResult<()> {
|
||||||
|
// Candidate application, memory revision, and intervention recording are
|
||||||
|
// one atomic transaction in the subjektiv Feature database.
|
||||||
|
self.database.transaction(|tx| {
|
||||||
|
tx.execute(
|
||||||
|
"INSERT INTO memories (subject_id, memory_id, body)
|
||||||
|
VALUES (?1, ?2, ?3)",
|
||||||
|
params![subject_id, memory_id, body],
|
||||||
|
)?;
|
||||||
|
tx.execute(
|
||||||
|
"UPDATE subjects SET revision = revision + 1
|
||||||
|
WHERE subject_id = ?1",
|
||||||
|
[subject_id],
|
||||||
|
)?;
|
||||||
|
tx.execute(
|
||||||
|
"INSERT INTO interventions (intervention_id, subject_id, detail)
|
||||||
|
VALUES (?1, ?2, 'candidate applied')",
|
||||||
|
params![intervention_id, subject_id],
|
||||||
|
)?;
|
||||||
|
Ok(())
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn initialize_subjektiv_repository(api: &WorkspaceApi) -> StorageResult<SubjektivRepository> {
|
||||||
|
let registered = api.feature_storage().register(REGISTRATION)?;
|
||||||
|
let database = api.feature_storage().open(®istered)?;
|
||||||
|
Ok(SubjektivRepository { database })
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
A production Feature should register once while constructing its Server service and retain the returned `RegisteredFeature`/typed repository. Duplicate registration is rejected. `FeatureDatabase` methods are synchronous because `rusqlite` is synchronous; async HTTP/background handlers must run nontrivial repository work on a blocking executor rather than blocking a Tokio core thread.
|
||||||
|
|
||||||
|
## Migration and transaction contract
|
||||||
|
|
||||||
|
- Versions are contiguous from 1. Migration schema changes and the corresponding history rows are committed in one `BEGIN IMMEDIATE` transaction.
|
||||||
|
- Concurrent opens in one manager are single-flight. Independent Server managers/processes serialize migration writers through SQLite and reread history after acquiring the transaction.
|
||||||
|
- A migration callback error rolls back the complete migration batch. A later open retries it; failure is never recorded as success.
|
||||||
|
- Stored migration names are verified. A history mismatch is rejected, and a schema newer than the registered plan is never overwritten or downgraded.
|
||||||
|
- `FeatureDatabase::transaction` covers multiple tables in one Feature database. No atomic transaction is promised with the Server control-plane database or another Feature database.
|
||||||
|
- A domain rejection can return `FeatureStorageError::operation(...)`; the transaction then rolls back.
|
||||||
|
|
||||||
|
## Lifecycle and operations
|
||||||
|
|
||||||
|
### Disable and restart
|
||||||
|
|
||||||
|
Not registering/opening a disabled Feature creates nothing. Disabling a previously used Feature does not delete its database. On restart, registering and opening the same Feature/Workspace reuses the persisted file and verifies/applies its migrations.
|
||||||
|
|
||||||
|
### Backup and restore
|
||||||
|
|
||||||
|
`WorkspaceFeatureStorage::backup` uses SQLite's online backup API for each discovered Feature database, so a live WAL is not ignored or copied as a bare database file. Each image passes `integrity_check` and `foreign_key_check`, and a manifest binds the snapshot to its Workspace and Feature identifiers. The snapshot directory is published with an atomic rename only after all images succeed.
|
||||||
|
|
||||||
|
Each Feature image is internally consistent. There is deliberately no cross-database snapshot instant or distributed transaction between Feature databases or the Server DB. A caller that needs a higher-level coordinated backup must first stop relevant Feature writers.
|
||||||
|
|
||||||
|
`restore` rejects the wrong Workspace manifest, corrupt/foreign-key-invalid images, an existing destination, or any live connection. It builds validated databases in a temporary directory and atomically publishes the Workspace storage directory. Restore is an operator/Server lifecycle operation, not a model API.
|
||||||
|
|
||||||
|
### Workspace deletion
|
||||||
|
|
||||||
|
Workspace deletion calls `WorkspaceFeatureStorage::delete` after Worker/Workdir cleanup and before finalizing removal from the Server DB. It fences new opens, waits for connection-locked operations, closes handles, and removes only the validated Workspace subtree. The admission fence remains until the cached Workspace API is discarded, preventing a background task from recreating storage between filesystem deletion and control-plane finalization. Retrying deletion is idempotent.
|
||||||
|
|
||||||
|
### Shutdown
|
||||||
|
|
||||||
|
After HTTP graceful drain, Server serve paths explicitly call `shutdown`. Shutdown rejects new registration/open, waits for in-progress connection operations to release their locks, closes all managed connections, and makes retained handles fail with `DatabaseClosed`. Operations must not treat a shutdown error as success.
|
||||||
|
|
||||||
|
## Failure handling
|
||||||
|
|
||||||
|
Identifier, registration, migration-history, unsupported-schema, maintenance, shutdown, snapshot, I/O, and SQLite failures are typed as `FeatureStorageError`. Server handlers should map them to a bounded domain/API error without exposing paths, SQL, or unrestricted database access. Migration/restore/delete failures must stop the corresponding lifecycle transition; they are not warnings to ignore.
|
||||||
Reference in New Issue
Block a user