feat(storage): serve typed usage reports from the daemon
Let the daemon count its own storage. Online clients can leave Python and SQLite at home.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155// Package daemon provides typed API access and local daemon lifecycle operations.package daemon
import ( "context" "encoding/json" "net/http" "time")
type StorageFile struct { Path string `json:"path"` Bytes int64 `json:"bytes"`}
type StorageSession struct { ID string `json:"id"` Bytes int64 `json:"bytes"`}
type StorageDatabase struct { Sessions []StorageSession `json:"sessions"` Images int64 `json:"images"` FreePages int64 `json:"free_pages"` PageSize int64 `json:"page_size"`}
type StorageReport struct { OldKernels []StorageFile `json:"old_kernels"` OldBackups []StorageFile `json:"old_backups"` DB StorageDatabase `json:"db"` Database int64 `json:"database"` WAL int64 `json:"wal"` Kernels int64 `json:"kernels"` Backups int64 `json:"backups"` Other int64 `json:"other"` RecentBackups int64 `json:"recent_backups"` RecentBackupCount int `json:"recent_backup_count"`}
func GetStorageReport(ctx context.Context, conn *Connection) (StorageReport, error) { if ctx == nil { ctx = context.Background() } ctx, cancel := context.WithTimeout(ctx, 20*time.Second) defer cancel() if err := checkCapability(ctx, conn, "storage_report", "for storage reports"); err != nil { return StorageReport{}, err } var result StorageReport err := executeRead(ctx, conn, operation{Name: "get storage report", Method: http.MethodGet, Path: "/storage/report", Policy: readRecovery}, func(data []byte) error { var wire struct { OldKernels []*storageFileWire `json:"old_kernels"` OldBackups []*storageFileWire `json:"old_backups"` DB *struct { Sessions []*struct { ID *string `json:"id"` Bytes *int64 `json:"bytes"` } `json:"sessions"` Images *int64 `json:"images"` FreePages *int64 `json:"free_pages"` PageSize *int64 `json:"page_size"` } `json:"db"` Database *int64 `json:"database"` WAL *int64 `json:"wal"` Kernels *int64 `json:"kernels"` Backups *int64 `json:"backups"` Other *int64 `json:"other"` RecentBackups *int64 `json:"recent_backups"` RecentBackupCount *int `json:"recent_backup_count"` } if err := json.Unmarshal(data, &wire); err != nil { return err } if wire.DB == nil { return fieldError("db") } for _, field := range []struct { name string value *int64 }{ {"database", wire.Database}, {"wal", wire.WAL}, {"kernels", wire.Kernels}, {"backups", wire.Backups}, {"other", wire.Other}, {"recent_backups", wire.RecentBackups}, {"images", wire.DB.Images}, {"free_pages", wire.DB.FreePages}, {"page_size", wire.DB.PageSize}, } { if field.value == nil || *field.value < 0 { return fieldError(field.name) } } if wire.RecentBackupCount == nil || *wire.RecentBackupCount < 0 { return fieldError("recent_backup_count") } kernels, err := decodeStorageFiles(wire.OldKernels, "old_kernels") if err != nil { return err } backups, err := decodeStorageFiles(wire.OldBackups, "old_backups") if err != nil { return err } if wire.DB.Sessions == nil { return fieldError("sessions") } sessions := make([]StorageSession, 0, len(wire.DB.Sessions)) for _, row := range wire.DB.Sessions { if row == nil || row.ID == nil || *row.ID == "" { return fieldError("id") } if row.Bytes == nil || *row.Bytes < 0 { return fieldError("bytes") } sessions = append(sessions, StorageSession{ID: *row.ID, Bytes: *row.Bytes}) } result = StorageReport{ OldKernels: kernels, OldBackups: backups, DB: StorageDatabase{ Sessions: sessions, Images: *wire.DB.Images, FreePages: *wire.DB.FreePages, PageSize: *wire.DB.PageSize, }, Database: *wire.Database, WAL: *wire.WAL, Kernels: *wire.Kernels, Backups: *wire.Backups, Other: *wire.Other, RecentBackups: *wire.RecentBackups, RecentBackupCount: *wire.RecentBackupCount, } return nil }) return result, err}
type storageFileWire struct { Path *string `json:"path"` Bytes *int64 `json:"bytes"`}
func decodeStorageFiles(rows []*storageFileWire, field string) ([]StorageFile, error) { if rows == nil { return nil, fieldError(field) } result := make([]StorageFile, 0, len(rows)) for _, row := range rows { if row == nil || row.Path == nil || *row.Path == "" { return nil, fieldError("path") } if row.Bytes == nil || *row.Bytes < 0 { return nil, fieldError("bytes") } result = append(result, StorageFile{Path: *row.Path, Bytes: *row.Bytes}) } return result, nil}12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182// Controlled peers can return malformed reports that a healthy daemon cannot// produce. These checks keep invalid sizes and missing arrays out of callers.package daemon
import ( "errors" "net" "net/http" "net/http/httptest" "strings" "testing")
const emptyStorageReport = `{"old_kernels":[],"old_backups":[],"db":{"sessions":[],"images":0,"free_pages":0,"page_size":0},"database":0,"wal":0,"kernels":0,"backups":0,"other":0,"recent_backups":0,"recent_backup_count":0}`
func TestStorageReportRejectsInvalidWireBeforeDelivery(t *testing.T) { for _, scenario := range []struct { name, body string status int valid bool }{ {"empty", emptyStorageReport, 200, true}, {"large integer", strings.Replace(emptyStorageReport, `"database":0`, `"database":9007199254740993`, 1), 200, true}, {"null report", `null`, 200, false}, {"missing database", strings.Replace(emptyStorageReport, `"database":0,`, "", 1), 200, false}, {"null size", strings.Replace(emptyStorageReport, `"database":0`, `"database":null`, 1), 200, false}, {"negative size", strings.Replace(emptyStorageReport, `"images":0`, `"images":-1`, 1), 200, false}, {"fraction", strings.Replace(emptyStorageReport, `"database":0`, `"database":1.0000000000000001`, 1), 200, false}, {"null files", strings.Replace(emptyStorageReport, `"old_kernels":[]`, `"old_kernels":null`, 1), 200, false}, {"null file entry", strings.Replace(emptyStorageReport, `"old_kernels":[]`, `"old_kernels":[null]`, 1), 200, false}, {"missing file bytes", strings.Replace(emptyStorageReport, `"old_kernels":[]`, `"old_kernels":[{"path":"kernel.state"}]`, 1), 200, false}, {"empty path", strings.Replace(emptyStorageReport, `"old_backups":[]`, `"old_backups":[{"path":"","bytes":1}]`, 1), 200, false}, {"null sessions", strings.Replace(emptyStorageReport, `"sessions":[]`, `"sessions":null`, 1), 200, false}, {"null session", strings.Replace(emptyStorageReport, `"sessions":[]`, `"sessions":[null]`, 1), 200, false}, {"empty session ID", strings.Replace(emptyStorageReport, `"sessions":[]`, `"sessions":[{"id":"","bytes":0}]`, 1), 200, false}, {"negative session size", strings.Replace(emptyStorageReport, `"sessions":[]`, `"sessions":[{"id":"s","bytes":-1}]`, 1), 200, false}, {"null count", strings.Replace(emptyStorageReport, `"recent_backup_count":0`, `"recent_backup_count":null`, 1), 200, false}, {"fractional count", strings.Replace(emptyStorageReport, `"recent_backup_count":0`, `"recent_backup_count":0.5`, 1), 200, false}, {"wrong status", emptyStorageReport, 202, false}, } { t.Run(scenario.name, func(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/health" { _, _ = w.Write([]byte(`{"ok":true,"version":2,"capabilities":["storage_report"]}`)) return } if r.URL.Path != "/storage/report" || r.Method != http.MethodGet { t.Errorf("unexpected request %s %s", r.Method, r.URL.Path) } w.WriteHeader(scenario.status) _, _ = w.Write([]byte(scenario.body)) })) defer server.Close() conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, nil) defer conn.HTTPClient().CloseIdleConnections() report, err := GetStorageReport(t.Context(), conn) if scenario.valid { if err != nil { t.Fatal(err) } if report.OldKernels == nil || report.OldBackups == nil || report.DB.Sessions == nil { t.Fatalf("missing empty arrays: %+v", report) } if scenario.name == "large integer" && report.Database != 9007199254740993 { t.Fatalf("rounded size: %d", report.Database) } } else { if err == nil { t.Fatalf("accepted invalid report: %+v", report) } if report.Database != 0 || report.OldKernels != nil || report.DB.Sessions != nil { t.Fatalf("returned partial report: %+v", report) } if scenario.status == 200 { if _, ok := errors.AsType[*ProtocolError](err); !ok { t.Fatalf("missing protocol failure: %v", err) } } } }) }}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596%% Filesystem accounting only. Never open SQLite or follow storage symlinks.-module(albedo_storage_report).-export([inspect/2]).-include_lib("kernel/include/file.hrl").
inspect(Home, Sessions) -> try directory(Home), Database = filename:join(Home, <<"albedo.sqlite">>), DbBytes = regular_size(Database), Wal = regular_size(<<Database/binary, "-wal">>) + regular_size(<<Database/binary, "-shm">>), %% Native mtimes have second precision. Strict comparison conservatively %% delays eligibility by less than a second instead of marking a newer file old. Cutoff = erlang:system_time(second) - 30 * 24 * 60 * 60, Known = sets:from_list(Sessions), {Kernels, OldKernels, _, _} = storage_files(filename:join(Home, <<"kernels">>), kernel, Known, Cutoff), {Backups, OldBackups, RecentBytes, RecentCount} = storage_files(filename:join(Home, <<"backups">>), backup, Known, Cutoff), Excluded = [<<"kernels">>, <<"backups">>, <<"albedo.sqlite">>, <<"albedo.sqlite-wal">>, <<"albedo.sqlite-shm">>], Other = lists:sum([size_if_regular(filename:join(Home, Name)) || Name <- names(Home), not lists:member(Name, Excluded)]), {ok, {filesystem, OldKernels, OldBackups, DbBytes, Wal, Kernels, Backups, Other, RecentBytes, RecentCount}} catch throw:{storage_error, Reason} -> {error, Reason} end.
directory(Path) -> case info(Path) of missing -> missing; #file_info{type = directory} -> ok; _ -> fail(Path, <<"expected a directory for storage">>) end.
regular_size(Path) -> case info(Path) of missing -> 0; #file_info{type = regular, size = Size} -> Size; _ -> fail(Path, <<"expected a regular file for storage">>) end.
size_if_regular(Path) -> case entry_info(Path) of #file_info{type = regular, size = Size} -> Size; _ -> 0 end.
info(Path) -> case file:read_link_info(Path, [{time, posix}]) of {ok, Info} -> Info; {error, enoent} -> missing; {error, Reason} -> fail(Path, atom_to_binary(Reason)) end.
entry_info(Path) -> case info(Path) of missing -> fail(Path, <<"entry disappeared during storage inspection">>); Info -> Info end.
has_suffix(Name, Suffix) -> byte_size(Name) >= byte_size(Suffix) andalso binary:part(Name, byte_size(Name) - byte_size(Suffix), byte_size(Suffix)) =:= Suffix.
names(Path) -> case file:list_dir(Path) of {ok, Names} -> lists:sort([unicode:characters_to_binary(Name) || Name <- Names]); {error, Reason} -> fail(Path, atom_to_binary(Reason)) end.
storage_files(Dir, Kind, Known, Cutoff) -> case directory(Dir) of missing -> {0, [], 0, 0}; ok -> {Total, Old, Recent, Count} = lists:foldl(fun(Name, {Bytes, Candidates, RecentBytes, RecentCount}) -> Path = filename:join(Dir, Name), case entry_info(Path) of #file_info{type = regular, size = Size, mtime = Mtime} -> Eligible = eligible(Kind, Name, Known), case {Eligible, Mtime < Cutoff, Kind} of {true, true, _} -> {Bytes + Size, [{file_bytes, Path, Size} | Candidates], RecentBytes, RecentCount}; {true, false, backup} -> {Bytes + Size, Candidates, RecentBytes + Size, RecentCount + 1}; _ -> {Bytes + Size, Candidates, RecentBytes, RecentCount} end; _ -> {Bytes, Candidates, RecentBytes, RecentCount} end end, {0, [], 0, 0}, names(Dir)), {Total, lists:reverse(Old), Recent, Count} end.
eligible(kernel, Name, Known) -> case has_suffix(Name, <<".state">>) of true -> not sets:is_element(binary:part(Name, 0, byte_size(Name) - 6), Known); false -> false end;eligible(backup, Name, _) -> string:prefix(Name, <<"albedo-before-image-store-">>) =/= nomatch andalso has_suffix(Name, <<".sqlite">>).
fail(Path, Reason) -> throw({storage_error, <<Reason/binary, ": ", Path/binary>>}).16 unmodified lines1718192021222324251976 unmodified linesExpand all2002200320042005200520062007200820092010255 unmodified linesExpand all22662267226822692270227122722273227419 unmodified lines2294229522962297229822992300230123022303230423052306230723081025 unmodified linesExpand all16 unmodified linesimport albedo/daemon/session_providerimport albedo/daemon/session_submissionimport albedo/daemon/settingsimport albedo/daemon/state_expiryimport albedo/daemon/storage_reportimport albedo/daemon/storeimport albedo/daemon/turnimport albedo/daemon/usageimport albedo/harness/cache_ttl1976 unmodified linesExpand all
/// The daemon's own top-level routes; a service never shadows them.const daemon_routes = [ "operations", "settings", "health", "sessions", "models", "auth", "shutdown", "agents", "quota", "cache-ttl", "fs", "hosts", "storage", "agents", "quota", "cache-ttl", "fs", "hosts",]
/// Shutdown can close a handle after the registry admitted its request./// Keep this transport failure at the HTTP boundary; other panics still fail.255 unmodified linesExpand all #( "capabilities", json.array( [ "storage_report", "operation_receipts", "session_stream_generation", "agents_stream_overflow", "submission_cancellation",19 unmodified lines } reply(200, acknowledged(fields)) } } Get, ["storage", "report"] -> case actor.call(registry, 5000, Host) { Error(message) -> error(503, message) Ok(host) -> storage_report.report(runtime.ledger(host), config.home) |> answered(200, fn(report) { report }, 500) } Get, ["settings"] -> settings.snapshot(config.home) |> answered_settings Put, ["settings", "providers", name] -> body(req, decode.dynamic) |> result.try(fn(value) {1025 unmodified linesExpand all123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161//// Read-only storage accounting. SQL stays on the store owner; filesystem//// inspection runs in the HTTP request process after the projection returns.
import albedo/daemon/storeimport gleam/dynamic/decodeimport gleam/jsonimport gleam/listimport gleam/resultimport sqlight
type SessionBytes { SessionBytes(id: String, bytes: Int)}
type Database { Database( sessions: List(SessionBytes), images: Int, free_pages: Int, page_size: Int, )}
pub type FileBytes { FileBytes(path: String, bytes: Int)}
pub type Filesystem { Filesystem( old_kernels: List(FileBytes), old_backups: List(FileBytes), database: Int, wal: Int, kernels: Int, backups: Int, other: Int, recent_backups: Int, recent_backup_count: Int, )}
@external(erlang, "albedo_storage_report", "inspect")fn inspect(home: String, sessions: List(String)) -> Result(Filesystem, String)
pub fn report(db: store.Store, home: String) -> Result(json.Json, String) { use database <- result.try(store.query(db, projection)) use files <- result.try(inspect( home, list.map(database.sessions, fn(session) { session.id }), )) Ok( json.object([ #("old_kernels", json.array(files.old_kernels, file_json)), #("old_backups", json.array(files.old_backups, file_json)), #( "db", json.object([ #( "sessions", json.array(database.sessions, fn(session) { json.object([ #("id", json.string(session.id)), #("bytes", json.int(session.bytes)), ]) }), ), #("images", json.int(database.images)), #("free_pages", json.int(database.free_pages)), #("page_size", json.int(database.page_size)), ]), ), #("database", json.int(files.database)), #("wal", json.int(files.wal)), #("kernels", json.int(files.kernels)), #("backups", json.int(files.backups)), #("other", json.int(files.other)), #("recent_backups", json.int(files.recent_backups)), #("recent_backup_count", json.int(files.recent_backup_count)), ]), )}
fn file_json(file: FileBytes) -> json.Json { json.object([ #("path", json.string(file.path)), #("bytes", json.int(file.bytes)), ])}
fn projection(db: sqlight.Connection) -> Result(Database, String) { use tables <- result.try( store.rows(db, "SELECT name FROM sqlite_master WHERE type='table'", [], { use value <- decode.field(0, decode.string) decode.success(value) }), ) use _ <- result.try( case list.contains(tables, "sessions") && list.contains(tables, "transcript") { True -> Ok(Nil) False -> Error("storage report requires the sessions and transcript tables") }, ) let transcript = "COALESCE((SELECT SUM(LENGTH(payload)) FROM transcript WHERE session=s.id),0)" let cells = case list.contains(tables, "cells") { False -> "0" True -> { let #(join, trace) = case list.contains(tables, "cell_traces") { True -> #( "LEFT JOIN cell_traces t ON t.id=c.id", "COALESCE(LENGTH(t.payload),0)", ) False -> #("", "0") } "COALESCE((SELECT SUM(LENGTH(c.source)+COALESCE(LENGTH(c.payload),0)+" <> trace <> ") FROM cells c " <> join <> " WHERE c.session=s.id),0)" } } use sessions <- result.try( store.rows( db, "SELECT s.id,COALESCE(LENGTH(s.pinned_context),0)+" <> transcript <> "+" <> cells <> " FROM sessions s", [], { use id <- decode.field(0, decode.string) use bytes <- decode.field(1, decode.int) decode.success(SessionBytes(id, bytes)) }, ), ) use images <- result.try(case list.contains(tables, "images") { True -> scalar(db, "SELECT COALESCE(SUM(LENGTH(data)),0) FROM images") False -> Ok(0) }) use free_pages <- result.try(scalar(db, "PRAGMA freelist_count")) use page_size <- result.try(scalar(db, "PRAGMA page_size")) Ok(Database(sessions, images, free_pages, page_size))}
fn scalar(db: sqlight.Connection, sql: String) -> Result(Int, String) { store.one( db, sql, [], { use value <- decode.field(0, decode.int) decode.success(value) }, "missing storage accounting value", )}