"terminal_size",

name = "env_filter"

version = "0.1.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "a009aa4810eb158359dda09d0c87378e4bbb89b5a801f016885a4707ba24f7ea"

dependencies = [

 "log",

 "regex",

]

[[package]]

name = "env_logger"

version = "0.11.3"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "38b35839ba51819680ba087cd351788c9a3c476841207e0b8cee0b04722343b9"

dependencies = [

 "anstream",

 "anstyle",

 "env_filter",

 "humantime",

 "log",

]

[[package]]

 "env_logger 0.10.2",

 "env_logger 0.11.3",

 "sqlite3-sys",

 "env_logger 0.10.2",

name = "terminal_size"

version = "0.3.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "21bebf2b7c9e0a515f6e0f8c51dc0f8e4696391e6f1ff30379559f8365fb0df7"

dependencies = [

 "rustix",

 "windows-sys 0.48.0",

]

[[package]]

clap = { version = "4.5.4", features = ["derive", "wrap_help"] }

sqlite3-sys = "0.15.0"

env_logger = "0.11.3"

~~~{#broker-allow-nothing.yaml .file .yaml}

db: ci-broker.db

report_dir: reports

default_adapter: mcadapterface

adapters:

  mcadapterface:

    command: ./adapter.sh

    env:

      RADICLE_NATIVE_CI: native-ci.yaml

    sensitive_env: {}

filters:

  - !Branch "this-branch-does-not-exist"

~~~

## A `refsFetched` node event

This is a node event from the node to signal that some git refs have

changed in repository.

~~~{#refsfetched.json .file .json}

        "name": "refs/namespaces/DUMMYNID/refs/heads/main",

## A shutdown event

## Set rid in `refsFetched` event

This is a helper script that reads a `refsFetched` event and changes

its `rid` field to be the id of the given repository. It also sets the

        oo["updated"]["name"] = nid.join(name.split("DUMMYNID"))

## Set environment variables and run command

To avoid having to repeat environment variables to set up and use a

Radicle node for verification scenarios, we provide a script that

sets them and runs a command.

~~~{#radenv.sh .file .sh}

#!/bin/bash

set -euo pipefail

homedir="$(pwd)/homedir"

env \

	HOME="$homedir" \

	RAD_PASSPHRASE=secret \

	RAD_HOME="$homedir/.radicle" \

	RAD_SOCKET=synt.sock \

	RUST_LOG=trace \

	RADICLE_CI_BROKER_LOG=trace \

	"$@"

~~~

## Set up a node with repository

Most of our verification scenarios will need to set up a Radicle node

and a test repository there. Rather than repeat it in every scenario,

we use this helper script.

~~~{#setup-node.sh .file .sh}

#!/bin/bash

set -xeuo pipefail

mkdir -p "$HOME"

env | grep RAD

rad auth --alias brokertest

git config --global user.email radicle@example.com

git config --global user.name TestyMcTestFace

git init -b main testy

cd testy

echo "test file" > test.txt

git add .

git commit -am test

git status

git show

rad init --name testy --description test --default-branch main --private --no-confirm --no-seed

rad inspect --identity

rad id list

~~~

## Shows config as JSON

_Requirement:_ The CI broker can write out the configuration is uses

at run time as JSON.

_Justification:_ This is helpful for the node operator to verify that

they have configured the program correctly.

_Stakeholder:_ Lars

Our verification here is quite simplistic, and only checks that the

output is in the JSON format. It does not try to make sure the JSON

matches the YAML semantically.

~~~scenario

given an installed cib

given file broker.yaml

when I run cib --config broker.yaml config --output actual.json

when I run jq . actual.json

then command is successful

~~~

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given file refsfetched.json

when I run env HOME=../homedir python3 set-rid refsfetched.json testy

when I run synthetic-events synt.sock refsfetched.json --log log.txt

given an installed cib

when I run bash radenv.sh RAD_SOCKET=synt.sock cib --config broker.yaml process-events

given an installed cibtool

when I run cibtool --db ci-broker.db event list

then stdout is exactly ""

when I run cibtool --db ci-broker.db run list

then stdout contains "id: "xyzzy""

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given file refsfetched.json

when I run env HOME=../homedir python3 set-rid refsfetched.json testy

when I run synthetic-events synt.sock refsfetched.json --log log.txt

given an installed cib

given file broker.yaml

given a directory reports

when I try to run bash radenv.sh RAD_SOCKET=synt.sock cib --config broker.yaml insert

then command is successful

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cib

when I try to run bash radenv.sh RAD_SOCKET=xyzzy.sock cib --config broker.yaml insert

then stderr contains "node control socket does not exist: xyzzy.sock"

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cib

when I try to run env HOME=homedir cib --config not-yaml.yaml config

then stderr contains "failed to parse configuration file as YAML: not-yaml.yaml"

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cib

when I try to run bash radenv.sh RAD_SOCKET=synt.sock cib --config broker.yaml insert

then command is successful

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cib

when I run synthetic-events synt.sock --log synt.log

when I try to run bash radenv.sh RAD_SOCKET=synt.sock cib --config broker.yaml insert

when I run cat synt.log

when I run cibtool --db ci-broker.db run list

then stdout is exactly ""

~~~

## Produces a report page upon request

_Requirement:_ The node operator can run a command to produce a report

of all CI runs a CI broker instance has performed.

_Justification:_ This is useful for diagnosis, if nothing else.

_Stakeholder:_ Lars.

~~~scenario

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cibtool

when I run cibtool --db x.db run add --id x --repo rad:zwTxygwuz5LDGBq255RA2CbNGrz8 --alias x --url https://x/1 --branch main --commit f1815dde6ae406d8fe3cec0b96c4486766342716 --who x --finished --failure --timestamp 2024-07-09T02:00:00

given a directory reports

when I run bash radenv.sh cibtool --db x.db report --output-dir reports

then file reports/index.html exists

then file reports/index.html contains "zwTxygwuz5LDGBq255RA2CbNGrz8"

then file reports/zwTxygwuz5LDGBq255RA2CbNGrz8.html exists

then file reports/zwTxygwuz5LDGBq255RA2CbNGrz8.html contains "success"

# Acceptance criteria for persistent database

The CI broker uses an SQLite database for persistent data. Many

processes may need to access or modify the database at the same time.

While SQLite is good at managing that, it needs to be used in the

right way for everything to work correctly. The acceptance criteria in

this chapter address that.

To enable the verification of these acceptance criteria, the CI broker

database allows for a "counter", as a single row in a dedicated table.

Concurrency is tested by having multiple processes update the counter

at the same time and verifying the end result is as intended and that

every value is set exactly once.

## Count in a single process

_Requirement:_ A single process can increment the test counter

correctly.

_Justification:_ If this doesn't work with a single process, it won't

work of multiple processes, either.

_Stakeholder:_ Lars.

~~~scenario

given an installed cibtool

then file count.db does not exist

when I run cibtool --db count.db counter show

then stdout is exactly "0\n"

when I run cibtool --db count.db counter count --goal 1000

when I run cibtool --db count.db counter show

then stdout is exactly "1000\n"

~~~

## Insert events into queue

_Requirement:_ Insert broker events generated from node events into

persistent event queue in the database, when allowed by the CI broker

event filter.

_Justification:_ This is fundamental for running CI when repositories

in a node change.

_Stakeholder:_ Lars.

~~~scenario

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cib

given an installed synthetic-events

given file refsfetched.json

when I run synthetic-events synt.sock refsfetched.json --log synt.log

given file broker.yaml

when I try to run bash radenv.sh env RAD_SOCKET=synt.sock cib --config broker.yaml insert

then command is successful

when I run cibtool --db ci-broker.db event list --verbose

then stdout contains "RefChanged"

then stdout contains "oid: Oid(0000000000000000000000000000000000000000)"

then stdout contains "old: Some(Oid(0000000000000000000000000000000000000000))"

~~~

## Insert many events into queue

_Requirement:_ Insert many events that arrive quickly.

_Justification:_ We need at least some rudimentary performance testing.

_Stakeholder:_ Lars.

~~~scenario

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cib

given an installed synthetic-events

given file refsfetched.json

when I run synthetic-events synt.sock refsfetched.json --log synt.log --repeat 1000

given file broker.yaml

when I try to run bash radenv.sh env RAD_SOCKET=synt.sock cib --config broker.yaml insert

then command is successful

when I run cibtool --db ci-broker.db event count

then stdout is exactly "1000\n"

~~~

## Don't insert events into queue when not allowed by filter

_Requirement:_ Nothing is inserted into the persistent event queue

then the CI broker's filter does not allow any events.

_Justification:_ This is fundamental for running CI when repositories

in a node change.

_Stakeholder:_ Lars.

~~~scenario

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cib

given an installed synthetic-events

given file refsfetched.json

when I run synthetic-events synt.sock refsfetched.json --log synt.log

given file broker.yaml from broker-allow-nothing.yaml

when I try to run bash radenv.sh env RAD_SOCKET=synt.sock cib --config broker.yaml insert

then command is successful

when I run cibtool --db ci-broker.db event count

then stdout is exactly "0\n"

~~~

## Process queued events

_Requirement:_ It's possible to run the CI broker in a mode where it

only processes events from its persistent event queue.

_Justification:_ This is primarily useful for testing the CI broker

queuing implementation.

_Stakeholders:_ Lars.

We verify this by adding events to the queue with `cibtool`, and then

running the CI broker and verifying it terminates after processing the

events. We carefully add a shutdown event so that the CI broker shuts

down.

~~~scenario

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given file adapter.sh from dummy.sh

when I run chmod +x adapter.sh

given an installed cib

given an installed cibtool

when I run bash radenv.sh cibtool --db ci-broker.db event add --repo testy --ref main --commit HEAD

when I run cibtool --db ci-broker.db event shutdown

given file broker.yaml

when I run ls -l adapter.sh

when I run bash radenv.sh cib --config broker.yaml queued

then stderr contains "Action: run:"

then stderr contains "Action: shutdown"

when I run cibtool --db ci-broker.db event list

then stdout is exactly ""

when I run cibtool --db ci-broker.db run list

then stdout contains "Success"

~~~

## Count in concurrent processes

_Requirement:_ Two process can concurrently increment the test counter

correctly.

_Justification:_ This is necessary, if not necessarily sufficient, for

concurrent database use to work correctly.

_Stakeholder:_ Lars.

Due to limitations in Subplot we mange the concurrent processes using

a helper shell script,k `count.sh`, found below. It runs two

concurrent `cibtool` processes that update the same database file, and

count to a desired goal. The script then verifies that everything went

correctly.

~~~scenario

given an installed cibtool

given file count.sh

when I run env RUST_LOG=debug bash -x count.sh 100 10

then stdout contains "OK\n"

~~~

~~~{#count.sh .file .sh}

#!/bin/bash

set -euo pipefail

run() {

	cibtool --db "$DB" counter count --goal "$goal"

}

DB=count.db

goal="$1"

reps="$2"

for x in $(seq "$reps"); do

	echo "Repetition $x"

	rm -f "$DB" ./?.out

	run >1.out 2>&1 &

	one=$!

	run >2.out 2>&1 &

	two=$!

	if ! wait "$one"; then

		echo "first run failed"

		cat 1.out

		exit 1

	fi

	if ! wait "$two"; then

		echo "second run failed"

		cat 2.out

		exit 1

	fi

	if grep ERROR ./?.out; then

		echo found ERRORs

		exit 1

	fi

	n="$(sqlite3 "$DB" 'select counter from counter_test')"

	[ "$n" == "$goal" ] || (

		echo "wrong count $n"

		exit 1

	)

	if awk '/increment to/ { print $NF }' ./?.out | sort -n | uniq -d | grep .; then

		echo "duplicate increments"

		exit 1

	fi

done

echo OK

~~~

# Acceptance criteria for event queue management

The CI builder database contains a queue of broker events. The

`cibtool` management tool can be used to examine and manipulate the

queue.

## Events can be queued and removed from queue

_Requirement:_ `cibtool` can show the queued events, can inject an

event, and remove an event.

_Justification:_ This is the minimum functionality needed to manage

the event queue.

_Stakeholder:_ Lars.

We verify that this works by adding a new broker event, and then

removing it. We randomly choose the repository id for the CI broker

itself for this test, but the id shouldn't matter, it just needs to

be of the correct form.

~~~scenario

given file radenv.sh

given file setup-node.sh

when I run bash radenv.sh bash setup-node.sh

given an installed cibtool

when I run cibtool --db x.db event list

then stdout is exactly ""

when I run bash radenv.sh cibtool --db x.db event add --repo testy --ref main --commit HEAD --base c0ffee --id-file id.txt

when I run cibtool --db x.db event show --id-file id.txt

then stdout contains "rad:"

then stdout contains "main"

then stdout contains "c0ffee"

when I run cibtool --db x.db event remove --id-file id.txt

when I run cibtool --db x.db event list

then stdout is exactly ""

~~~

## Can add shutdown event to queue

_Requirement:_ `cibtool` can add a shutdown event to the queued

events.

_Justification:_ This is needed for testing, and for the node operator

to be able to do this cleanly.

_Stakeholder:_ Lars.

~~~scenario

given an installed cibtool

when I run cibtool --db x.db event list

then stdout is exactly ""

when I run cibtool --db x.db event shutdown --id-file id.txt

when I run cibtool --db x.db event show --id-file id.txt

then stdout contains "Shutdown"

~~~

## Add CI run information to database

_Requirement:_ `cibtool` can add information about a CI run, possibly

one that is imaginary.

_Justification:_ This is primarily needed for testing.

_Stakeholder:_ Lars.

~~~scenario

given an installed cibtool

when I run cibtool --db x.db run list

then stdout is exactly ""

when I run cibtool --db x.db run add --id x --repo rad:zwTxygwuz5LDGBq255RA2CbNGrz8 --alias x --url https://x/1 --branch main --commit f1815dde6ae406d8fe3cec0b96c4486766342716 --who x --finished --failure --timestamp 2024-07-09T02:00:00

when I run cibtool --db x.db run list

then stdout contains "rad:zwTxygwuz5LDGBq255RA2CbNGrz8"

~~~

- given: "an installed cib"

  impl:

    rust:

      function: install_cib

- given: "an installed cibtool"

  impl:

    rust:

      function: install_cibtool

    pub fn run(&self, trigger: &Request, run: &mut Run) -> Result<(), AdapterError> {

        let x = self.run_helper(trigger, run);

    fn run_helper(&self, trigger: &Request, run: &mut Run) -> Result<(), AdapterError> {

        debug!("spawn adapter sub-process: {:?}", self.bin);

    use super::{Adapter, Run};

        Adapter::new(&bin).run(&trigger_request()?, &mut run)?;

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

        let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);

//! The CI broker.

#![allow(clippy::result_large_err)]

use std::{

    error::Error,

    fs::write,

    path::{Path, PathBuf},

    process::exit,

};

use clap::Parser;

use log::{debug, error, info};

use radicle::Profile;

use radicle_ci_broker::{

    adapter::Adapter,

    broker::{Broker, BrokerError},

    config::{Config, ConfigError},

    db::{Db, DbError},

    pages::{PageBuilder, PageError},

    queueadd::{AdderError, QueueAdderBuilder},

    queueproc::{QueueError, QueueProcessorBuilder},

};

fn main() {

    if let Err(e) = fallible_main() {

        error!("ERROR: {}", e);

        let mut e = e.source();

        while let Some(source) = e {

            error!("caused by: {}", source);

            e = source.source();

        }

        exit(1);

    }

}

fn fallible_main() -> Result<(), CibError> {

    let args = Args::parse();

    pretty_env_logger::init_custom_env("RADICLE_CI_BROKER_LOG");

    info!("Radicle CI broker starts");

    let config = Config::load(&args.config).map_err(|e| CibError::read_config(&args.config, e))?;

    debug!("loaded configuration: {:#?}", config);

    args.run(&config)?;

    Ok(())

}

#[derive(Debug, Parser)]

struct Args {

    #[clap(long)]

    config: PathBuf,

    #[clap(subcommand)]

    cmd: Cmd,

}

impl Args {

    fn run(&self, config: &Config) -> Result<(), CibError> {

        match &self.cmd {

            Cmd::Config(x) => x.run(self, config)?,

            Cmd::Insert(x) => x.run(self, config)?,

            Cmd::Queued(x) => x.run(self, config)?,

            Cmd::ProcessEvents(x) => x.run(self, config)?,

        }

        Ok(())

    }

    fn open_db(&self, config: &Config) -> Result<Db, CibError> {

        Db::new(&config.db).map_err(CibError::db)

    }

}

#[derive(Debug, Parser)]

enum Cmd {

    Config(ConfigCmd),

    Insert(InsertCmd),

    Queued(QueuedCmd),

    ProcessEvents(ProcessEventsCmd),

}

#[derive(Debug, Parser)]

struct ConfigCmd {

    #[clap(long)]

    output: Option<PathBuf>,

}

impl ConfigCmd {

    fn run(&self, _args: &Args, config: &Config) -> Result<(), CibError> {

        let json = config.to_json();

        if let Some(output) = &self.output {

            write(output, json.as_bytes()).map_err(|e| CibError::write_config(output, e))?;

        } else {

            println!("{json}");

        }

        Ok(())

    }

}

#[derive(Debug, Parser)]

struct InsertCmd {}

impl InsertCmd {

    fn run(&self, args: &Args, config: &Config) -> Result<(), CibError> {

        let adder = QueueAdderBuilder::default()

            .db(args.open_db(config)?)

            .filters(&config.filters)

            .build()

            .map_err(CibError::QueueAdder)?;

        let thread = adder.add_events_in_thread();

        thread

            .join()

            .expect("wait for thread to finish")

            .map_err(CibError::add_events)?;

        debug!("cib ends");

        Ok(())

    }

}

#[derive(Debug, Parser)]

struct QueuedCmd {}

impl QueuedCmd {

    fn run(&self, args: &Args, config: &Config) -> Result<(), CibError> {

        let db = args.open_db(config)?;

        let mut broker = Broker::new(config.db()).map_err(CibError::new_broker)?;

        let spec =

            config

                .adapter(&config.default_adapter)

                .ok_or(CibError::UnknownDefaultAdapter(

                    config.default_adapter.clone(),

                ))?;

        let adapter = Adapter::new(&spec.command)

            .with_environment(spec.envs())

            .with_environment(spec.sensitive_envs());

        debug!("default adapter: {adapter:?}");

        broker.set_default_adapter(&adapter);

        let processor = QueueProcessorBuilder::default()

            .db(db)

            .broker(broker)

            .build()

            .map_err(CibError::process_queue)?;

        let thread = processor.process_in_thread();

        thread

            .join()

            .expect("wait for thread to finish")

            .map_err(CibError::process_queue)?;

        debug!("cib ends");

        Ok(())

    }

}

#[derive(Debug, Parser)]

struct ProcessEventsCmd {}

impl ProcessEventsCmd {

    fn run(&self, args: &Args, config: &Config) -> Result<(), CibError> {

        let adder = QueueAdderBuilder::default()

            .db(args.open_db(config)?)

            .filters(&config.filters)

            .push_shutdown()

            .build()

            .map_err(CibError::QueueAdder)?;

        let adder = adder.add_events_in_thread();

        let profile = Profile::load().map_err(CibError::profile)?;

        let db = args.open_db(config)?;

        let mut page = PageBuilder::default()

            .node_alias(&profile.config.node.alias)

            .runs(db.get_all_runs().map_err(CibError::db)?)

            .build()

            .map_err(CibError::page_updater)?;

        if let Some(dirname) = &config.report_dir {

            page.set_output_dir(dirname);

        }

        let thread = page.update_in_thread(db, &profile.config.node.alias, true);

        thread.join().unwrap().map_err(CibError::page_updater)?;

        let mut broker = Broker::new(config.db()).map_err(CibError::new_broker)?;

        let spec =

            config

                .adapter(&config.default_adapter)

                .ok_or(CibError::UnknownDefaultAdapter(

                    config.default_adapter.clone(),

                ))?;

        let adapter = Adapter::new(&spec.command)

            .with_environment(spec.envs())

            .with_environment(spec.sensitive_envs());

        debug!("default adapter: {adapter:?}");

        broker.set_default_adapter(&adapter);

        let processor = QueueProcessorBuilder::default()

            .db(args.open_db(config)?)

            .broker(broker)

            .build()

            .map_err(CibError::process_queue)?;

        let processor = processor.process_in_thread();

        processor

            .join()

            .expect("wait for processor thread to finish")

            .map_err(CibError::process_queue)?;

        adder

            .join()

            .expect("wait for adder thread to finish")

            .map_err(CibError::add_events)?;

        debug!("cib ends");

        Ok(())

    }

}

#[derive(Debug, thiserror::Error)]

enum CibError {

    #[error("failed to read configuration file {0}")]

    ReadConfig(PathBuf, #[source] ConfigError),

    #[error("failed to write config as JSON to file {0}")]

    WriteConfig(PathBuf, #[source] std::io::Error),

    #[error("failed to look up node profile")]

    Profile(#[source] radicle::profile::Error),

    #[error("failed to use SQLite database")]

    Db(#[source] DbError),

    #[error("failed to create report page")]

    PageUpdater(#[source] PageError),

    #[error("failed create broker data type")]

    NewBroker(#[source] BrokerError),

    #[error("failed to add node events into queue")]

    QueueAdder(#[source] AdderError),

    #[error("failed to process events from queue")]

    ProcessQueue(#[source] QueueError),

    #[error("failed to add events to queue")]

    AddEvents(#[source] AdderError),

    #[error("default adapter is not in list of adapters")]

    UnknownDefaultAdapter(String),

}

impl CibError {

    fn read_config(filename: &Path, e: ConfigError) -> Self {

        Self::ReadConfig(filename.into(), e)

    }

    fn write_config(filename: &Path, e: std::io::Error) -> Self {

        Self::WriteConfig(filename.into(), e)

    }

    fn profile(e: radicle::profile::Error) -> Self {

        Self::Profile(e)

    }

    fn db(e: DbError) -> Self {

        Self::Db(e)

    }

    fn page_updater(e: PageError) -> Self {

        Self::PageUpdater(e)

    }

    fn new_broker(e: BrokerError) -> Self {

        Self::NewBroker(e)

    }

    fn process_queue(e: QueueError) -> Self {

        Self::ProcessQueue(e)

    }

    fn add_events(e: AdderError) -> Self {

        Self::AddEvents(e)