Skip to content

Preserve observer custody and avoid projection churn #798

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
flip-engineer merged 1 commit into from
Oct 17, 2026
Merged

Preserve observer custody and avoid projection churn #798

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
182 changes: 170 additions & 12 deletions bend2/src/host/process-spawn.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
#include <fcntl.h>
#include <spawn.h>
#include <sys/wait.h>
#include <dirent.h>

extern char **environ;

Expand Down Expand Up @@ -1213,6 +1214,131 @@ static int br_observer_pid(int socket,pid_t *pid) {
*pid=peer.pid;return 0;
#endif
}
/* A live replacement must identify the supplied process from the exact retained
attempt it is observing. The pidfd remains open through handoff and retirement. */
static int br_observer_attempt_custody(pid_t pid,const char *database,const char *session,
const char *directory,int lock,int *life_out) {
if(!life_out)return EINVAL;
*life_out=-1;
char *canonical=realpath(directory,NULL),*database_path=realpath(database,NULL);
if(!canonical || !database_path) {free(canonical);free(database_path);return errno;}
char *stdout_path=br_path(canonical,"stdout");
char *log_path=br_path(canonical,"observer.log");
int error=!stdout_path || !log_path?ENOMEM:0;
int ended=0,life=-1;
if(!error) {
life=br_process_lifetime(pid,&ended);
if(life<0)error=ended?ESRCH:errno;
}
#ifndef __APPLE__
char command_path[PATH_MAX];
unsigned char command[65536];
size_t command_length=0;
if(!error) {
int n=snprintf(command_path,sizeof(command_path),"/proc/%ld/cmdline",(long)pid);
if(n<0 || (size_t)n>=sizeof(command_path))error=EOVERFLOW;
}
if(!error) {
int fd=open(command_path,O_RDONLY|O_CLOEXEC);
if(fd<0)error=errno;
else {
while(command_length<sizeof(command)) {
ssize_t n=read(fd,command+command_length,sizeof(command)-command_length);
if(n<0 && errno==EINTR)continue;
if(n<0){error=errno;break;}
if(!n)break;
command_length+=(size_t)n;
}
if(!error && command_length==sizeof(command))error=E2BIG;
close(fd);
}
}
if(!error) {
char *argv[16];size_t argc=0,start=0;
if(!command_length || command[command_length-1])error=EINVAL;
else {
for(size_t i=0;i<command_length;i++)if(!command[i]) {
if(argc==sizeof(argv)/sizeof(argv[0])){error=E2BIG;break;}
argv[argc++]=(char *)command+start;start=i+1;
}
int receive=!error && (argc==14 || argc==15) && !strcmp(argv[1],"--recover-receive");
int recovery=!error && argc==5 && !strcmp(argv[2],"recover-observer");
if(!error && !receive && !recovery)error=EPERM;
if(!error && strcmp(argv[3],session))error=EPERM;
if(!error) {
char *saved_database=realpath(receive?argv[2]:argv[1],NULL);
if(!saved_database || strcmp(saved_database,database_path))error=EPERM;
free(saved_database);
}
if(!error && receive) {
char *saved_directory=realpath(argv[12],NULL);
if(!saved_directory || strcmp(saved_directory,canonical))error=EPERM;
free(saved_directory);
const char *name=strrchr(canonical,'/');name=name?name+1:canonical;
const char *encoded=strstr(name,".attempt-");
const char *id=argv[4];static const char hex[]="0123456789abcdef";
size_t id_length=strlen(id);
if(!encoded || strlen(encoded+9)!=id_length*2)error=EPERM;
else for(size_t i=0;i<id_length;i++) {
unsigned char c=(unsigned char)id[i];
if(encoded[9+i*2]!=hex[c>>4] || encoded[10+i*2]!=hex[c&15]){error=EPERM;break;}
}
}
if(!error && recovery) {
char *end=NULL;errno=0;unsigned long prior=strtoul(argv[4],&end,10);
if(errno || end==argv[4] || *end || !prior || prior>INT_MAX)error=EPERM;
}
}
}

if(!error) {
struct pollfd alive={life,POLLIN,0};
int ready;do {ready=poll(&alive,1,0);}while(ready<0 && errno==EINTR);
if(ready<0)error=errno;
else if(ready || (alive.revents&(POLLIN|POLLHUP|POLLERR)))error=ESRCH;
}
if(!error) {
struct stat lock_info;
if(fstat(lock,&lock_info))error=errno;
char fd_directory[PATH_MAX],fd_path[PATH_MAX],target[PATH_MAX+1];
int length=snprintf(fd_directory,sizeof(fd_directory),"/proc/%ld/fd",(long)pid);
if(!error && (length<0 || (size_t)length>=sizeof(fd_directory)))error=EOVERFLOW;
DIR *fds=!error?opendir(fd_directory):NULL;
if(!error && !fds)error=errno;
int stdout_owned=0,log_owned=0,lock_owned=0;
struct dirent *entry;
while(!error && (entry=readdir(fds))) {
if(entry->d_name[0]=='.')continue;
length=snprintf(fd_path,sizeof(fd_path),"%s/%s",fd_directory,entry->d_name);
if(length<0 || (size_t)length>=sizeof(fd_path)){error=EOVERFLOW;break;}
ssize_t n=readlink(fd_path,target,sizeof(target)-1);
if(n>=0) {
target[n]=0;
if(!strcmp(target,stdout_path))stdout_owned=1;
if(!strcmp(target,log_path))log_owned=1;
}
struct stat info;
if(!stat(fd_path,&info) && info.st_dev==lock_info.st_dev && info.st_ino==lock_info.st_ino)lock_owned=1;
if(stdout_owned && log_owned && lock_owned)break;
}
if(fds)closedir(fds);
if(!error && (!stdout_owned || !log_owned || !lock_owned))error=EPERM;
}
if(!error) {
struct pollfd alive={life,POLLIN,0};
int ready;do {ready=poll(&alive,1,0);}while(ready<0 && errno==EINTR);
if(ready<0)error=errno;
else if(ready || (alive.revents&(POLLIN|POLLHUP|POLLERR)))error=ESRCH;
}
#else
if(!error)error=ENOTSUP;
#endif
free(stdout_path);free(log_path);free(canonical);free(database_path);
if(error) {if(life>=0)close(life);return error;}
*life_out=life;
return 0;
}

static int br_control_command(BrKeeper *keeper,BrControl *control,BrFrame frame,const char *payload) {
const char *end=memchr(payload,0,(size_t)frame.length);
int error=0;
Expand Down Expand Up @@ -4969,7 +5095,7 @@ static void __attribute__((constructor)) baton_process_signals(void){signal(SIGP
typedef struct {
char *database, *session, *path, *directory;
const char *detail;
int handle, error, kind, keeper;
int handle, error, kind, keeper, observer_life;
u32 observer;
BatonProcessCall attempt;
} BatonSessionLock;
Expand All @@ -4979,19 +5105,38 @@ static int baton_observer_completed(BatonSessionLock *call) {
int error=br_admission_verify(call->directory,call->database,-1);
if(error)return error;
char *path=br_path(call->directory,"status");
FILE *file=path?fopen(path,"r"):NULL;
error=file?0:path?errno:ENOMEM;
if(!path)return ENOMEM;
errno=0;
FILE *file=fopen(path,"r");
int status_error=file?0:errno;
free(path);
int status=0;
int status=0,live_pending=0;
if(file) {
if(fscanf(file,"%d",&status)!=1)error=EINVAL;
else if(!WIFEXITED(status) && !WIFSIGNALED(status))error=EBUSY;
if(fscanf(file,"%d",&status)!=1)status_error=EINVAL;
else if(!WIFEXITED(status) && !WIFSIGNALED(status))status_error=EBUSY;
fclose(file);
} else if(status_error==ENOENT) {
live_pending=1;
status_error=0;
}
if(error) {call->detail="Observer recovery requires the retained native exit status.";return error;}
if(status_error) {call->detail="Observer recovery requires a valid retained native status or a live native birth.";return status_error;}
BrBirth native;
if((error=br_read_file(call->directory,"native.birth",&native,sizeof(native))))return error;
if(call->observer==(u32)native.pid)return EINVAL;
if(live_pending) {
int ended=0,life=br_lifetime(call->directory,&ended);
if(life<0) {
error=ended?ESRCH:errno;
call->detail="Live observer recovery requires the recorded native birth to remain alive.";
return error;
}
close(life);
error=br_observer_attempt_custody((pid_t)call->observer,call->database,call->session,call->directory,call->handle,&call->observer_life);
if(error) {
call->detail="Live observer recovery requires the supplied process to hold this attempt's stdout and observer log.";
return error;
}
}
BrManifest manifest={0};
if((error=br_manifest_read(call->directory,&manifest))) {
br_manifest_free(&manifest);return error;
Expand All @@ -5006,25 +5151,36 @@ static int baton_observer_completed(BatonSessionLock *call) {
call->keeper=1;
} else {
error=errno;
if(error==ENOENT || error==ECONNREFUSED)error=0;
if(!live_pending && (error==ENOENT || error==ECONNREFUSED))error=0;
}
}
if(probe>=0)close(probe);
if(!error && live_pending && !call->keeper) {
error=ECONNREFUSED;
call->detail="Live observer recovery requires the active native keeper.";
}
return error;
}

static int baton_observer_retire(BatonSessionLock *call) {
int ended=0,life=br_process_lifetime((pid_t)call->observer,&ended);
if(life<0 && !ended)return errno;
int ended=0,life=call->observer_life;
call->observer_life=-1;
if(life<0)life=br_process_lifetime((pid_t)call->observer,&ended);
if(life<0)return ended?0:errno;
int error=0;
#ifdef SYS_pidfd_send_signal
if(syscall(SYS_pidfd_send_signal,life,SIGTERM,NULL,0) && errno!=ESRCH)error=errno;
if(!error && syscall(SYS_pidfd_send_signal,life,SIGCONT,NULL,0) && errno!=ESRCH)error=errno;
#else
if(kill((pid_t)call->observer,SIGTERM) && errno!=ESRCH)error=errno;
if(!error && kill((pid_t)call->observer,SIGCONT) && errno!=ESRCH)error=errno;
if(!error && life>=0) {
#endif
if(!error) {
struct pollfd observed={life,POLLIN,0};
int ready;do {ready=poll(&observed,1,-1);}while(ready<0 && errno==EINTR);
if(ready<0)error=errno;
}
if(life>=0)close(life);
close(life);
return error;
}

Expand Down Expand Up @@ -5146,6 +5302,7 @@ static Term baton_session_lock_pack(Env e, IoWork *w) {
}
}
Term result=call->error ? io_fail(e,call->error,call->detail) : io_done(e,value);
if(call->observer_life>=0)close(call->observer_life);
free(call->database);free(call->session);free(call->path);free(call->directory);free(call);
w->data=NULL;
return result;
Expand All @@ -5155,6 +5312,7 @@ static Term baton_session_lock_begin(Env e, Term *f, IoWork *w, int kind) {
BatonSessionLock *call=calloc(1,sizeof(*call));
if(!call) return io_fail(e,ENOMEM,NULL);
call->kind=kind;
call->observer_life=-1;
if(kind==3 || kind==5 || kind==6) {
u64 dn=0,sn=0;
call->database=io_cstr(e,f[0],&dn);
Expand Down
70 changes: 68 additions & 2 deletions bend2/src/host/sqlite.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,61 @@ static const char *const baton_change_schema[] = {
NULL
};

/* Return whether a DROP TRIGGER is immediately followed by the same trigger
definition already stored in sqlite_schema. SQLite omits the final semicolon
from sqlite_schema.sql, so compare after trimming that terminator. */
static int baton_sql_trigger_matches(sqlite3 *db, const char *drop_sql,
const char *create_sql, int *matches,
char **error) {
static const char drop_prefix[] = "DROP TRIGGER IF EXISTS ";
static const char create_prefix[] = "CREATE TRIGGER ";
*matches = 0;
if (strncmp(drop_sql, drop_prefix, sizeof(drop_prefix) - 1) != 0 ||
strncmp(create_sql, create_prefix, sizeof(create_prefix) - 1) != 0)
return SQLITE_OK;
const char *drop_name = drop_sql + sizeof(drop_prefix) - 1;
size_t drop_name_length = strcspn(drop_name, "; \t\r\n");
const char *create_name = create_sql + sizeof(create_prefix) - 1;
size_t create_name_length = strcspn(create_name, " \t\r\n");
if (!drop_name_length || drop_name[drop_name_length] != ';' ||
drop_name_length != create_name_length ||
memcmp(drop_name, create_name, drop_name_length) != 0)
return SQLITE_OK;

sqlite3_stmt *statement = NULL;
int code = sqlite3_prepare_v2(db,
"SELECT sql FROM sqlite_master WHERE type='trigger' AND name=?1;",
-1, &statement, NULL);
if (code == SQLITE_OK)
code = sqlite3_bind_text(statement, 1, drop_name, (int)drop_name_length, SQLITE_STATIC);
if (code == SQLITE_OK) {
code = sqlite3_step(statement);
if (code == SQLITE_ROW) {
const char *stored = (const char *)sqlite3_column_text(statement, 0);
size_t stored_length = (size_t)sqlite3_column_bytes(statement, 0);
size_t wanted_length = strlen(create_sql);
while (stored_length && (stored[stored_length - 1] == ';' ||
stored[stored_length - 1] == ' ' || stored[stored_length - 1] == '\t' ||
stored[stored_length - 1] == '\r' || stored[stored_length - 1] == '\n'))
stored_length--;
while (wanted_length && (create_sql[wanted_length - 1] == ';' ||
create_sql[wanted_length - 1] == ' ' || create_sql[wanted_length - 1] == '\t' ||
create_sql[wanted_length - 1] == '\r' || create_sql[wanted_length - 1] == '\n'))
wanted_length--;
*matches = stored && stored_length == wanted_length &&
memcmp(stored, create_sql, wanted_length) == 0;
code = SQLITE_OK;
} else if (code == SQLITE_DONE) {
code = SQLITE_OK;
}
}
int final = sqlite3_finalize(statement);
if (code == SQLITE_OK && final != SQLITE_OK) code = final;
if (code != SQLITE_OK && error && !*error)
*error = sqlite3_mprintf("%s", sqlite3_errmsg(db));
return code;
}

static int baton_sql_install_projection(sqlite3 *db, const char *schema, char **error) {
int code = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, error);
if (code == SQLITE_OK)
Expand All @@ -74,8 +129,19 @@ static int baton_sql_install_projection(sqlite3 *db, const char *schema, char **
code = sqlite3_exec(db, "ALTER TABLE knowledge_promotions ADD COLUMN source_kind TEXT NOT NULL DEFAULT 'session';", NULL, NULL, error);
if (code == SQLITE_OK && sqlite3_table_column_metadata(db, NULL, "knowledge_promotions", "destination_kind", NULL, NULL, NULL, NULL, NULL) != SQLITE_OK)
code = sqlite3_exec(db, "ALTER TABLE knowledge_promotions ADD COLUMN destination_kind TEXT NOT NULL DEFAULT 'session';", NULL, NULL, error);
for (size_t i = 0; code == SQLITE_OK && baton_change_schema[i]; i++)
code = sqlite3_exec(db, baton_change_schema[i], NULL, NULL, error);
for (size_t i = 0; code == SQLITE_OK && baton_change_schema[i]; i++) {
int matches = 0;
if (baton_change_schema[i + 1] &&
strncmp(baton_change_schema[i], "DROP TRIGGER IF EXISTS ", 23) == 0)
code = baton_sql_trigger_matches(db, baton_change_schema[i],
baton_change_schema[i + 1], &matches, error);
if (code == SQLITE_OK && matches) {
i++;
continue;
}
if (code == SQLITE_OK)
code = sqlite3_exec(db, baton_change_schema[i], NULL, NULL, error);
}
if (code == SQLITE_OK)
code = sqlite3_exec(db, "COMMIT", NULL, NULL, error);
if (code != SQLITE_OK)
Expand Down