fix: error when upload failed before all bytes received

Reviewed-on: vieter-v/vieter#355

src/agent/daemon.v

@@ -20,11 +20,13 @@ struct AgentDaemon {
 	client client.Client
 mut:
 	images ImageManager
-	// Which builds are currently running; length is conf.max_concurrent_builds
-	builds []BuildConfig
 	// Atomic variables used to detect when a build has finished; length is
-	// conf.max_concurrent_builds
+	// conf.max_concurrent_builds. This approach is used as the difference
+	// between a recently finished build and an empty build slot is important
+	// for knowing whether the agent is currently "active".
 	atomics []u64
+	// Channel used to send builds to worker threads
+	build_channel chan BuildConfig
 }
 
 // agent_init initializes a new agent
@@ -34,8 +36,8 @@ fn agent_init(logger log.Log, conf Config) AgentDaemon {
 		client: client.new(conf.address, conf.api_key)
 		conf: conf
 		images: new_image_manager(conf.image_rebuild_frequency * 60)
-		builds: []BuildConfig{len: conf.max_concurrent_builds}
 		atomics: []u64{len: conf.max_concurrent_builds}
+		build_channel: chan BuildConfig{cap: conf.max_concurrent_builds}
 	}
 
 	return d
@@ -43,6 +45,11 @@ fn agent_init(logger log.Log, conf Config) AgentDaemon {
 
 // run starts the actual agent daemon. This function will run forever.
 pub fn (mut d AgentDaemon) run() {
+	// Spawn worker threads
+	for builder_index in 0 .. d.conf.max_concurrent_builds {
+		spawn d.builder_thread(d.build_channel, builder_index)
+	}
+
 	// This is just so that the very first time the loop is ran, the jobs are
 	// always polled
 	mut last_poll_time := time.now().add_seconds(-d.conf.polling_frequency)
@@ -107,10 +114,10 @@ pub fn (mut d AgentDaemon) run() {
 				// It's technically still possible that the build image is
 				// removed in the very short period between building the
 				// builder image and starting a build container with it. If
-				// this happens, faith really just didn't want you to do this
+				// this happens, fate really just didn't want you to do this
 				// build.
 
-				d.start_build(config)
+				d.build_channel <- config
 				running++
 			}
 		}
@@ -147,22 +154,6 @@ fn (mut d AgentDaemon) update_atomics() (int, int) {
 	return finished, empty
 }
 
-// start_build starts a build for the given BuildConfig.
-fn (mut d AgentDaemon) start_build(config BuildConfig) bool {
-	for i in 0 .. d.atomics.len {
-		if stdatomic.load_u64(&d.atomics[i]) == agent.build_empty {
-			stdatomic.store_u64(&d.atomics[i], agent.build_running)
-			d.builds[i] = config
-
-			spawn d.run_build(i, config)
-
-			return true
-		}
-	}
-
-	return false
-}
-
 // run_build actually starts the build process for a given target.
 fn (mut d AgentDaemon) run_build(build_index int, config BuildConfig) {
 	d.linfo('started build: ${config}')
@@ -195,3 +186,12 @@ fn (mut d AgentDaemon) run_build(build_index int, config BuildConfig) {
 
 	stdatomic.store_u64(&d.atomics[build_index], agent.build_done)
 }
+
+// builder_thread is a thread that constantly listens for builds to process
+fn (mut d AgentDaemon) builder_thread(ch chan BuildConfig, builder_index int) {
+	for {
+		build_config := <-ch or { break }
+
+		d.run_build(builder_index, build_config)
+	}
+}

src/server/repo.v

@@ -68,7 +68,7 @@ fn (mut app App) put_package(repo_ string) web.Result {
 		mut sw := time.new_stopwatch(time.StopWatchOptions{ auto_start: true })
 
 		util.reader_to_file(mut app.reader, length.int(), pkg_path) or {
-			app.lwarn("Failed to upload '${pkg_path}'")
+			app.lwarn("Failed to upload '${pkg_path}': ${err.msg()}")
 
 			return app.status(.internal_server_error)
 		}

src/util/stream.v

@@ -46,6 +46,10 @@ pub fn reader_to_file(mut reader io.BufferedReader, length int, path string) ! {
 			to_write = to_write - bytes_written
 		}
 	}
+
+	if bytes_left > 0 {
+		return error('Not all bytes were received.')
+	}
 }
 
 // match_array_in_array[T] returns how many elements of a2 overlap with a1. For