capy-bearer/scripts/test_task_queue_native.cpp

#include "src/lib/types.cpp"
#include "src/lib/dvalue.cpp"
#include <fcntl.h>
#include <sys/wait.h>
#include <unistd.h>

String random_bytes(u64 count) { String out(count,0);int fd=open("/dev/urandom",O_RDONLY|O_CLOEXEC);if(fd<0)return "";size_t at=0;while(at<count){ssize_t n=read(fd,&out[at],count-at);if(n<=0)break;at+=(size_t)n;}close(fd);out.resize(at);return out; }
#include "src/lib/task_queue.cpp"

using namespace task_queue;
static Limits limits() { return {32,4096,0,32,1024,1000}; }
static String lease(char c) { return String(64,c); }
static void need(bool value,const char* name,bool& ok) { if(!value) { fprintf(stderr,"FAIL: %s\n",name);ok=false; } }
static bool id_ok(const String& id) { if(id.size()!=64)return false;for(char c:id)if(!((c>='0'&&c<='9')||(c>='a'&&c<='f')))return false;return true; }
static bool state(const Result&r,const String&s,bool done) { const DValue* actual=r.status.key("state"),*complete=r.status.key("done");return r.ok()&&actual&&complete&&actual->to_string()==s&&complete->to_bool()==done; }
static u64 task_files(const String& dir) { DIR*d=opendir(dir.c_str());if(!d)return 0;u64 n=0;while(dirent*e=readdir(d))if(String(e->d_name).ends_with(".task"))++n;closedir(d);return n; }
int main(int argc,char** argv) {
	if(argc==4&&String(argv[1])=="--submit") { DValue props;props["child"]=argv[3];return submit(argv[2],limits(),"TASK:CHILD",props).ok()?0:1; }
	bool ok=true;char root_template[]="/tmp/capy-task-queue.XXXXXX";char* made=mkdtemp(root_template);if(!made)return 2;String root=made;
	DValue props;props["name"]="first";props["nested"]["value"]="copied";Result first=submit(root,limits(),"TASK:ONE",props);need(first.ok()&&id_ok(first.task.id),"exact opaque ID",ok);need(state(status(root,limits(),first.task.id),"queued",false),"queued taxonomy and done",ok);
	Result claimed=claim(root,limits(),lease('a'));need(claimed.ok()&&claimed.task.id==first.task.id&&claimed.task.props["name"].to_string()=="first"&&claimed.task.props["nested"]["value"].to_string()=="copied","claim copies nested props once",ok);need(state(status(root,limits(),first.task.id),"running",false),"running taxonomy",ok);
	need(cancel(root,limits(),first.task.id).ok(),"cancel running request",ok);Result requests=cancellation_requests(root,limits(),1);need(requests.ok()&&requests.count==1&&requests.status["requests"][first.task.id].to_string()==lease('a'),"bounded cancellation enumeration",ok);need(succeed(root,limits(),first.task.id,lease('a')).ok()&&state(status(root,limits(),first.task.id),"canceled",true),"cancel race resolves canceled",ok);
	Result missing=status(root,limits(),lease('9'));need(state(missing,"missing",false),"missing taxonomy is status not error",ok);
	Result failed=submit(root,limits(),"TASK:FAIL",DValue("2"));Result failed_claim=claim(root,limits(),lease('b'));need(failed_claim.ok()&&fail(root,limits(),failed.task.id,lease('b'),"handler_failed").ok()&&state(status(root,limits(),failed.task.id),"failed",true),"failure is terminal without implicit retry",ok);
	Result crash=submit(root,limits(),"TASK:CRASH",DValue());set_claim_crash_after_rename_for_test(true);need(claim(root,limits(),lease('d')).code=="injected_crash","claim fault after rename",ok);set_claim_crash_after_rename_for_test(false);need(task_files(root+"/pending")==0&&task_files(root+"/running")==1&&state(status(root,limits(),crash.task.id),"running",false),"crash leaves only empty-lease running record",ok);need(recover_worker(root,limits(),lease('d')).count==1&&state(status(root,limits(),crash.task.id),"failed",true)&&status(root,limits(),crash.task.id).status["failure_code"].to_string()=="worker_lost","recover empty lease as terminal worker loss",ok);String retained_root=root+"-retention";Result retained=submit(retained_root,limits(),"TASK:RETAIN",DValue());Result retained_claim=claim(retained_root,limits(),lease('a'));need(retained_claim.ok()&&succeed(retained_root,limits(),retained.task.id,lease('a')).ok(),"complete retention fixture",ok);DValue retained_status=status(retained_root,limits(),retained.task.id).status;retained_status["created_ms"]="1";retained_status["completed_ms"]=now_ms();String retained_data=brb_encode(retained_status);int retained_fd=open((retained_root+"/status/"+retained.task.id+".task").c_str(),O_WRONLY|O_TRUNC);need(retained_fd>=0&&write(retained_fd,retained_data.data(),retained_data.size())==(ssize_t)retained_data.size()&&fsync(retained_fd)==0&&close(retained_fd)==0,"set old creation retention fixture",ok);Limits retain=limits();retain.retention_seconds=1;need(reap(retained_root,retain).ok()&&state(status(retained_root,retain,retained.task.id),"succeeded",true),"retention uses completion not creation",ok);
	String malformed=root+"/pending/00000000000000000000-"+String(64,'f')+".task";int bad=open(malformed.c_str(),O_WRONLY|O_CREAT|O_TRUNC,0600);write(bad,"bad",3);close(bad);Result after_bad=submit(root,limits(),"TASK:AFTER",DValue("ok"));Result after_claim=claim(root,limits(),lease('f'));need(after_claim.ok()&&after_claim.task.id==after_bad.task.id,"malformed pending does not starve claim",ok);need(state(status(root,limits(),String(64,'f')),"failed",true)&&status(root,limits(),String(64,'f')).status["failure_code"].to_string()=="malformed_record","malformed is failed taxonomy",ok);if(after_claim.ok())need(succeed(root,limits(),after_claim.task.id,lease('f')).ok(),"complete after malformed",ok);
	Result await_task=submit(root,limits(),"TASK:AWAIT",DValue());need(state(await(root,limits(),await_task.task.id,5,1),"queued",false),"await expiry returns current status",ok);need(cancel(root,limits(),await_task.task.id).ok()&&state(status(root,limits(),await_task.task.id),"canceled",true),"queued cancel terminal",ok);
	String concurrent=root+"-concurrent";std::vector<pid_t> children;for(int n=0;n<6;n++){pid_t pid=fork();if(pid==0){String number=std::to_string(n);execl(argv[0],argv[0],"--submit",concurrent.c_str(),number.c_str(),(char*)0);_exit(127);}children.push_back(pid);}for(pid_t pid:children){int st=0;waitpid(pid,&st,0);need(WIFEXITED(st)&&WEXITSTATUS(st)==0,"concurrent process submit",ok);}std::set<String> ids;for(int n=0;n<6;n++){Result c=claim(concurrent,limits(),lease((char)('0'+n)));need(c.ok()&&ids.insert(c.task.id).second,"concurrent no duplicate claims",ok);if(c.ok())need(succeed(concurrent,limits(),c.task.id,lease((char)('0'+n))).ok(),"concurrent completion",ok);}
	usleep(2000);Result reaped=reap(root,limits());need(reaped.ok()&&reaped.count>=5,"retention uses terminal completion",ok);Limits batch=limits();batch.capacity=1;batch.scan_max=1;u64 batch_reaped=0;for(int n=0;n<6;n++)batch_reaped+=reap(concurrent,batch).count;need(batch_reaped==6,"bounded retention batches",ok);return ok?0:1;
}