uqbar-dao/nectar
1
1
Fork 0
mirror of https://github.com/uqbar-dao/nectar.git synced 2024-12-27 18:42:17 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into da/metadata

Browse Source
This commit is contained in:
tadad 2024-02-15 10:53:38 -06:00 committed by GitHub
commit 6af15a0f26
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 399 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
kinode/packages/app_store/app_store/src/http_api.rs
View File

@ -189,8 +189,38 @@ fn serve_paths(
}
Method::PUT => {
// update an app
// TODO
Ok((StatusCode::NO_CONTENT, None, format!("TODO").into_bytes()))
let pkg_listing: &PackageListing = state
.get_listing(&package_id)
.ok_or(anyhow::anyhow!("No package"))?;
let pkg_state: &PackageState = state
.downloaded_packages
.get(&package_id)
.ok_or(anyhow::anyhow!("No package"))?;
let download_from = pkg_state
.mirrored_from
.as_ref()
.ok_or(anyhow::anyhow!("No mirror for package {package_id}"))?
.to_string();
match crate::start_download(
our,
requested_packages,
&package_id,
&download_from,
pkg_state.mirroring,
pkg_state.auto_update,
&None,
) {
DownloadResponse::Started => Ok((
StatusCode::CREATED,
None,
format!("Downloading").into_bytes(),
)),
DownloadResponse::Failure => Ok((
StatusCode::SERVICE_UNAVAILABLE,
None,
format!("Failed to download").into_bytes(),
)),
}
}
Method::DELETE => {
// uninstall an app
@ -236,22 +266,34 @@ fn serve_paths(
}
Method::POST => {
// download an app
// TODO get fields from POST body
let pkg_listing: &PackageListing = state
.get_listing(&package_id)
.ok_or(anyhow::anyhow!("No package"))?;
// from POST body, look for download_from field and use that as the mirror
let body = crate::get_blob()
.ok_or(anyhow::anyhow!("missing blob"))?
.bytes;
let body_json: serde_json::Value =
serde_json::from_slice(&body).unwrap_or_default();
let mirrors: &Vec<NodeId> = pkg_listing
.metadata
.as_ref()
.expect("Package does not have metadata")
.properties
.mirrors
.as_ref();
// TODO select on FE
let download_from = mirrors
.first()
.as_ref()
.ok_or(anyhow::anyhow!("No mirrors for package {package_id}"))?;
// TODO select on FE
let download_from = body_json
.get("download_from")
.unwrap_or(&json!(mirrors
.first()
.ok_or(anyhow::anyhow!("No mirrors for package {package_id}"))?))
.as_str()
.ok_or(anyhow::anyhow!("download_from not a string"))?
.to_string();
// TODO select on FE? or after download but before install?
let mirror = false;
let auto_update = false;
let desired_version_hash = None;
@ -259,7 +301,7 @@ fn serve_paths(
our,
requested_packages,
&package_id,
download_from,
&download_from,
mirror,
auto_update,
&desired_version_hash,

43
kinode/packages/app_store/app_store/src/lib.rs
View File

@ -178,7 +178,7 @@ fn handle_message(
if source.node() != our.node() || source.process != "eth:distro:sys" {
return Err(anyhow::anyhow!("eth sub event from weird addr: {source}"));
}
handle_eth_sub_event(&mut state, e)?;
handle_eth_sub_event(our, &mut state, e)?;
}
Req::Http(incoming) => {
if source.node() != our.node()
@ -187,7 +187,7 @@ fn handle_message(
return Err(anyhow::anyhow!("http_server from non-local node"));
}
if let HttpServerRequest::Http(req) = incoming {
http_api::handle_http_request(&our, &mut state, requested_packages, &req)?;
http_api::handle_http_request(our, &mut state, requested_packages, &req)?;
}
}
},
@ -270,6 +270,7 @@ fn handle_local_request(
installed: false,
verified: true, // side loaded apps are implicitly verified because there is no "source" to verify against
caps_approved: true, // TODO see if we want to auto-approve local installs
manifest_hash: None, // generated in the add fn
mirroring: *mirror,
auto_update: false, // can't auto-update a local package
metadata: None, // TODO
@ -414,7 +415,7 @@ fn handle_receive_download(
);
} else {
return Err(anyhow::anyhow!(
"downloaded package is not latest version--rejecting download!"
"app store: downloaded package is not desired version--rejecting download! download hash: {download_hash}, desired hash: {hash}"
));
}
} else {
@ -449,7 +450,7 @@ fn handle_receive_download(
);
} else {
return Err(anyhow::anyhow!(
"downloaded package is not latest version--rejecting download!"
"app store: downloaded package is not latest version--rejecting download! download hash: {download_hash}, latest hash: {latest_hash}"
));
}
} else {
@ -458,6 +459,14 @@ fn handle_receive_download(
}
}
let old_manifest_hash = match state.downloaded_packages.get(&package_id) {
Some(package_state) => package_state
.manifest_hash
.clone()
.unwrap_or("OLD".to_string()),
_ => "OLD".to_string(),
};
state.add_downloaded_package(
&package_id,
PackageState {
@ -466,12 +475,28 @@ fn handle_receive_download(
installed: false,
verified,
caps_approved: false,
manifest_hash: None, // generated in the add fn
mirroring: requested_package.mirror,
auto_update: requested_package.auto_update,
metadata: None, // TODO
},
Some(blob.bytes),
)
)?;
let new_manifest_hash = match state.downloaded_packages.get(&package_id) {
Some(package_state) => package_state
.manifest_hash
.clone()
.unwrap_or("NEW".to_string()),
_ => "NEW".to_string(),
};
// lastly, if auto_update is true, AND the caps_hash has NOT changed,
// trigger install!
if requested_package.auto_update && old_manifest_hash == new_manifest_hash {
handle_install(our, state, &package_id)?;
}
Ok(())
}
fn handle_ft_worker_result(body: &[u8], context: &[u8]) -> anyhow::Result<()> {
@ -493,11 +518,15 @@ fn handle_ft_worker_result(body: &[u8], context: &[u8]) -> anyhow::Result<()> {
Ok(())
}
fn handle_eth_sub_event(state: &mut State, event: EthSubEvent) -> anyhow::Result<()> {
fn handle_eth_sub_event(
our: &Address,
state: &mut State,
event: EthSubEvent,
) -> anyhow::Result<()> {
let EthSubEvent::Log(log) = event else {
return Err(anyhow::anyhow!("app store: got non-log event"));
};
state.ingest_listings_contract_event(log)
state.ingest_listings_contract_event(our, log)
}
fn fetch_package_manifest(package: &PackageId) -> anyhow::Result<Vec<kt::PackageManifestEntry>> {

49
kinode/packages/app_store/app_store/src/types.rs
View File

@ -1,3 +1,4 @@
use crate::LocalRequest;
use alloy_rpc_types::Log;
use alloy_sol_types::{sol, SolEvent};
use kinode_process_lib::kernel_types as kt;
@ -72,6 +73,7 @@ pub struct PackageState {
pub installed: bool,
pub verified: bool,
pub caps_approved: bool,
pub manifest_hash: Option<String>,
/// are we serving this package to others?
pub mirroring: bool,
/// if we get a listing data update, will we try to download it?
@ -157,7 +159,7 @@ impl State {
pub fn add_downloaded_package(
&mut self,
package_id: &PackageId,
package_state: PackageState,
mut package_state: PackageState,
package_bytes: Option<Vec<u8>>,
) -> anyhow::Result<()> {
if let Some(package_bytes) = package_bytes {
@ -201,6 +203,13 @@ impl State {
})?)
.blob(blob)
.send_and_await_response(5)??;
let manifest_file = vfs::File {
path: format!("/{}/pkg/manifest.json", package_id),
};
let manifest_bytes = manifest_file.read()?;
let manifest_hash = generate_metadata_hash(&manifest_bytes);
package_state.manifest_hash = Some(manifest_hash);
}
self.downloaded_packages
.insert(package_id.to_owned(), package_state);
@ -282,6 +291,10 @@ impl State {
// generate entry from this data
// for the version hash, take the SHA-256 hash of the zip file
let our_version = generate_version_hash(&zip_file_bytes);
let manifest_file = vfs::File {
path: format!("/{}/pkg/manifest.json", package_id),
};
let manifest_bytes = manifest_file.read()?;
// the user will need to turn mirroring and auto-update back on if they
// have to reset the state of their app store for some reason. the apps
// themselves will remain on disk unless explicitly deleted.
@ -293,6 +306,7 @@ impl State {
installed: true,
verified: true, // implicity verified
caps_approved: true, // since it's already installed this must be true
manifest_hash: Some(generate_metadata_hash(&manifest_bytes)),
mirroring: false,
auto_update: false,
metadata: None,
@ -349,7 +363,11 @@ impl State {
}
/// only saves state if last_saved_block is more than 1000 blocks behind
pub fn ingest_listings_contract_event(&mut self, log: Log) -> anyhow::Result<()> {
pub fn ingest_listings_contract_event(
&mut self,
our: &Address,
log: Log,
) -> anyhow::Result<()> {
let block_number: u64 = log
.block_number
.ok_or(anyhow::anyhow!("app store: got log with no block number"))?
@ -458,6 +476,33 @@ impl State {
current_listing.metadata_hash = metadata_hash;
current_listing.metadata = metadata;
let package_id = PackageId::new(&current_listing.name, &current_listing.publisher);
// if we have this app installed, and we have auto_update set to true,
// we should try to download new version from the mirrored_from node
// and install it if successful.
if let Some(package_state) = self.downloaded_packages.get(&package_id) {
if package_state.auto_update {
if let Some(mirrored_from) = &package_state.mirrored_from {
crate::print_to_terminal(
1,
&format!(
"app store: auto-updating package {package_id} from {mirrored_from}"
),
);
Request::to(our)
.body(serde_json::to_vec(&LocalRequest::Download {
package: package_id,
download_from: mirrored_from.clone(),
mirror: package_state.mirroring,
auto_update: package_state.auto_update,
desired_version_hash: None,
})?)
.send()?;
}
}
}
}
Transfer::SIGNATURE_HASH => {
let from = alloy_primitives::Address::from_word(log.topics[1]);

90
kinode/packages/app_store/pkg/ui/assets/index-A09g5OKk.js Normal file
View File

File diff suppressed because one or more lines are too long

230
kinode/src/kernel/process.rs
View File

@ -112,11 +112,7 @@ impl ProcessState {
) -> Result<(wit::Address, wit::Message), (wit::SendError, Option<wit::Context>)> {
let res = match self.message_queue.pop_front() {
Some(message_from_queue) => message_from_queue,
None => self
.recv_in_process
.recv()
.await
.expect("fatal: process couldn't receive next message"),
None => self.ingest_message().await,
};
self.kernel_message_to_process_receive(res)
}
@ -138,11 +134,7 @@ impl ProcessState {
}
// next, wait for the awaited message to arrive
loop {
let res = self
.recv_in_process
.recv()
.await
.expect("fatal: process couldn't receive next message");
let res = self.ingest_message().await;
let id = match &res {
Ok(km) => km.id,
Err(e) => e.id,
@ -155,6 +147,131 @@ impl ProcessState {
}
}
/// ingest next valid message from kernel.
/// cancel any timeout task associated with this message.
/// if the message is a response, only enqueue if we have an outstanding request for it.
async fn ingest_message(&mut self) -> Result<t::KernelMessage, t::WrappedSendError> {
loop {
let message = self
.recv_in_process
.recv()
.await
.expect("fatal: process couldn't receive next message");
match &message {
Ok(km) => match &km.message {
t::Message::Response(_) => {
if let Some((_context, timeout_handle)) = self.contexts.get_mut(&km.id) {
timeout_handle.abort();
return message;
}
}
_ => {
return message;
}
},
Err(e) => {
if let Some((_context, timeout_handle)) = self.contexts.get_mut(&e.id) {
timeout_handle.abort();
return message;
}
}
}
}
}
/// Convert a message from the main event loop into a result for the process to receive.
/// If the message is a response or error, get context if we have one.
fn kernel_message_to_process_receive(
&mut self,
incoming: Result<t::KernelMessage, t::WrappedSendError>,
) -> Result<(wit::Address, wit::Message), (wit::SendError, Option<wit::Context>)> {
let (mut km, context) = match incoming {
Ok(mut km) => match km.message {
t::Message::Request(_) => {
self.last_blob = km.lazy_load_blob;
km.lazy_load_blob = None;
self.prompting_message = Some(km.clone());
(km, None)
}
t::Message::Response(_) => match self.contexts.remove(&km.id) {
Some((context, _timeout_handle)) => {
self.last_blob = km.lazy_load_blob;
km.lazy_load_blob = None;
self.prompting_message = context.prompting_message;
(km, context.context)
}
None => {
self.last_blob = km.lazy_load_blob;
km.lazy_load_blob = None;
self.prompting_message = Some(km.clone());
(km, None)
}
},
},
Err(e) => match self.contexts.remove(&e.id) {
None => return Err((t::en_wit_send_error(e.error), None)),
Some((context, _timeout_handle)) => {
self.prompting_message = context.prompting_message;
return Err((t::en_wit_send_error(e.error), context.context));
}
},
};
let pk = signature::UnparsedPublicKey::new(
&signature::ED25519,
self.keypair.as_ref().public_key(),
);
// prune any invalid capabilities before handing to process
// where invalid = supposedly issued by us, but not signed properly by us
match &mut km.message {
t::Message::Request(request) => {
request.capabilities.retain(|(cap, sig)| {
// The only time we verify a cap's signature is when a foreign node
// sends us a cap that we (allegedly) issued
if km.source.node != self.metadata.our.node
&& cap.issuer.node == self.metadata.our.node
{
match pk.verify(&rmp_serde::to_vec(&cap).unwrap_or_default(), sig) {
Ok(_) => true,
Err(_) => false,
}
} else {
return true;
}
});
}
t::Message::Response((response, _)) => {
response.capabilities.retain(|(cap, sig)| {
// The only time we verify a cap's signature is when a foreign node
// sends us a cap that we (allegedly) issued
if km.source.node != self.metadata.our.node
&& cap.issuer.node == self.metadata.our.node
{
match pk.verify(&rmp_serde::to_vec(&cap).unwrap_or_default(), sig) {
Ok(_) => true,
Err(_) => false,
}
} else {
return true;
}
});
}
};
Ok((
km.source.en_wit(),
match km.message {
t::Message::Request(request) => wit::Message::Request(t::en_wit_request(request)),
// NOTE: we throw away whatever context came from the sender, that's not ours
t::Message::Response((response, _sent_context)) => {
wit::Message::Response((t::en_wit_response(response), context))
}
},
))
}
/// takes Request generated by a process and sends it to the main event loop.
/// will only fail if process does not have capability to send to target.
/// if the request has a timeout (expects response), start a task to track
@ -362,99 +479,6 @@ impl ProcessState {
.await
.expect("fatal: kernel couldn't send response");
}
/// Convert a message from the main event loop into a result for the process to receive.
/// If the message is a response or error, get context if we have one.
fn kernel_message_to_process_receive(
&mut self,
incoming: Result<t::KernelMessage, t::WrappedSendError>,
) -> Result<(wit::Address, wit::Message), (wit::SendError, Option<wit::Context>)> {
let (mut km, context) = match incoming {
Ok(mut km) => match km.message {
t::Message::Request(_) => {
self.last_blob = km.lazy_load_blob;
km.lazy_load_blob = None;
self.prompting_message = Some(km.clone());
(km, None)
}
t::Message::Response(_) => {
if let Some((context, timeout_handle)) = self.contexts.remove(&km.id) {
timeout_handle.abort();
self.last_blob = km.lazy_load_blob;
km.lazy_load_blob = None;
self.prompting_message = context.prompting_message;
(km, context.context)
} else {
self.last_blob = km.lazy_load_blob;
km.lazy_load_blob = None;
self.prompting_message = Some(km.clone());
(km, None)
}
}
},
Err(e) => match self.contexts.remove(&e.id) {
None => return Err((t::en_wit_send_error(e.error), None)),
Some((context, timeout_handle)) => {
timeout_handle.abort();
self.prompting_message = context.prompting_message;
return Err((t::en_wit_send_error(e.error), context.context));
}
},
};
let pk = signature::UnparsedPublicKey::new(
&signature::ED25519,
self.keypair.as_ref().public_key(),
);
// prune any invalid capabilities before handing to process
// where invalid = supposedly issued by us, but not signed properly by us
match &mut km.message {
t::Message::Request(request) => {
request.capabilities.retain(|(cap, sig)| {
// The only time we verify a cap's signature is when a foreign node
// sends us a cap that we (allegedly) issued
if km.source.node != self.metadata.our.node
&& cap.issuer.node == self.metadata.our.node
{
match pk.verify(&rmp_serde::to_vec(&cap).unwrap_or_default(), sig) {
Ok(_) => true,
Err(_) => false,
}
} else {
return true;
}
});
}
t::Message::Response((response, _)) => {
response.capabilities.retain(|(cap, sig)| {
// The only time we verify a cap's signature is when a foreign node
// sends us a cap that we (allegedly) issued
if km.source.node != self.metadata.our.node
&& cap.issuer.node == self.metadata.our.node
{
match pk.verify(&rmp_serde::to_vec(&cap).unwrap_or_default(), sig) {
Ok(_) => true,
Err(_) => false,
}
} else {
return true;
}
});
}
};
Ok((
km.source.en_wit(),
match km.message {
t::Message::Request(request) => wit::Message::Request(t::en_wit_request(request)),
// NOTE: we throw away whatever context came from the sender, that's not ours
t::Message::Response((response, _sent_context)) => {
wit::Message::Response((t::en_wit_response(response), context))
}
},
))
}
}
/// create a specific process, and generate a task that will run it.

77
kinode/src/timer.rs
View File

@ -1,7 +1,7 @@
use anyhow::Result;
use lib::types::core::{
Address, KernelMessage, Message, MessageReceiver, MessageSender, PrintSender, Printout,
Response, TIMER_PROCESS_ID,
Response, TimerAction, TIMER_PROCESS_ID,
};
use serde::{Deserialize, Serialize};
@ -11,8 +11,8 @@ use serde::{Deserialize, Serialize};
/// requests made by other nodes.
///
/// The interface of the timer module is as follows:
/// One kind of request is accepted: the IPC must be a little-endian byte-representation
/// of an unsigned 64-bit integer, in milliseconds. This request should always expect a Response.
/// One kind of request is accepted: TimerAction::SetTimer(u64), where the u64 is the time to wait
/// in milliseconds. This request should always expect a Response.
/// If the request does not expect a Response, the timer will not be set.
///
/// A proper Request will trigger the timer module to send a Response. The Response will be
@ -39,44 +39,53 @@ pub async fn timer_service(
// we only handle Requests which contain a little-endian u64 as IPC,
// except for a special "debug" message, which prints the current state
let Message::Request(req) = km.message else { continue };
if req.body == "debug".as_bytes() {
let Ok(timer_action) = serde_json::from_slice::<TimerAction>(&req.body) else {
let _ = print_tx.send(Printout {
verbosity: 0,
content: format!("timer service active timers ({}):", timer_map.timers.len()),
verbosity: 1,
content: "timer service received a request with an invalid body".to_string(),
}).await;
for (k, v) in timer_map.timers.iter() {
continue
};
match timer_action {
TimerAction::Debug => {
let _ = print_tx.send(Printout {
verbosity: 0,
content: format!("{}: {:?}", k, v),
content: format!("timer service active timers ({}):", timer_map.timers.len()),
}).await;
for (k, v) in timer_map.timers.iter() {
let _ = print_tx.send(Printout {
verbosity: 0,
content: format!("{}: {:?}", k, v),
}).await;
}
continue
}
TimerAction::SetTimer(timer_millis) => {
// if the timer is set to pop in 0 millis, we immediately respond
// otherwise, store in our persisted map, and spawn a task that
// sleeps for the given time, then sends the response
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let pop_time = now + timer_millis;
if timer_millis == 0 {
send_response(&our, km.id, km.rsvp.unwrap_or(km.source), &kernel_message_sender).await;
continue
}
let _ = print_tx.send(Printout {
verbosity: 1,
content: format!("set timer to pop in {}ms", timer_millis),
}).await;
if !timer_map.contains(pop_time) {
timer_tasks.spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(timer_millis - 1)).await;
pop_time
});
}
timer_map.insert(pop_time, km.id, km.rsvp.unwrap_or(km.source));
}
continue
}
let Ok(bytes): Result<[u8; 8], _> = req.body.try_into() else { continue };
let timer_millis = u64::from_le_bytes(bytes);
// if the timer is set to pop in 0 millis, we immediately respond
// otherwise, store in our persisted map, and spawn a task that
// sleeps for the given time, then sends the response
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let pop_time = now + timer_millis;
if timer_millis == 0 {
send_response(&our, km.id, km.rsvp.unwrap_or(km.source), &kernel_message_sender).await;
continue
}
let _ = print_tx.send(Printout {
verbosity: 1,
content: format!("set timer to pop in {}ms", timer_millis),
}).await;
if !timer_map.contains(pop_time) {
timer_tasks.spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(timer_millis - 1)).await;
pop_time
});
}
timer_map.insert(pop_time, km.id, km.rsvp.unwrap_or(km.source));
}
Some(Ok(time)) = timer_tasks.join_next() => {
// when a timer pops, we send the response to the process(es) that set

6
lib/src/core.rs
View File

@ -1496,3 +1496,9 @@ impl From<tokio::sync::mpsc::error::SendError<CapMessage>> for SqliteError {
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TimerAction {
Debug,
SetTimer(u64),
}