diff --git a/PVE/Jobs.pm b/PVE/Jobs.pm
new file mode 100644
index 0000000..40caf25
--- /dev/null
+++ b/PVE/Jobs.pm
@@ -0,0 +1,380 @@
+package PVE::Jobs;
+
+use strict;
+use warnings;
+use JSON;
+
+use PVE::Cluster qw(cfs_lock_file cfs_read_file cfs_register_file);
+use PVE::Job::Registry;
+use PVE::Jobs::VZDump;
+use PVE::Jobs::RealmSync;
+use PVE::Tools;
+
+PVE::Jobs::VZDump->register();
+PVE::Jobs::RealmSync->register();
+PVE::Job::Registry->init();
+
+cfs_register_file(
+    'jobs.cfg',
+    sub { PVE::Job::Registry->parse_config(@_); },
+    sub { PVE::Job::Registry->write_config(@_); },
+);
+
+my $state_dir = "/var/lib/pve-manager/jobs";
+my $lock_dir = "/var/lock/pve-manager";
+
+my $get_state_file = sub {
+    my ($jobid, $type) = @_;
+    return "$state_dir/$type-$jobid.json";
+};
+
+my $default_state = {
+    state => 'created',
+    time => 0,
+};
+
+my $saved_config_props = [qw(enabled schedule)];
+
+# saves some properties of the jobcfg into the jobstate so we can track
+# them on different nodes (where the update was not done)
+# and update the last runtime when they change
+sub detect_changed_runtime_props {
+    my ($jobid, $type, $cfg) = @_;
+
+    lock_job_state(
+        $jobid,
+        $type,
+        sub {
+            my $old_state = read_job_state($jobid, $type) // $default_state;
+
+            my $updated = 0;
+            for my $prop (@$saved_config_props) {
+                my $old_prop = $old_state->{config}->{$prop} // '';
+                my $new_prop = $cfg->{$prop} // '';
+                next if "$old_prop" eq "$new_prop";
+
+                if (defined($cfg->{$prop})) {
+                    $old_state->{config}->{$prop} = $cfg->{$prop};
+                } else {
+                    delete $old_state->{config}->{$prop};
+                }
+
+                $updated = 1;
+            }
+
+            return if !$updated;
+            $old_state->{updated} = time();
+
+            my $path = $get_state_file->($jobid, $type);
+            PVE::Tools::file_set_contents($path, encode_json($old_state));
+        },
+    );
+}
+
+# lockless, since we use file_get_contents, which is atomic
+sub read_job_state {
+    my ($jobid, $type) = @_;
+    my $path = $get_state_file->($jobid, $type);
+    return if !-e $path;
+
+    my $raw = PVE::Tools::file_get_contents($path);
+
+    return $default_state if $raw eq '';
+
+    # untaint $raw
+    if ($raw =~ m/^(\{.*\})$/) {
+        return decode_json($1);
+    }
+
+    die "invalid json data in '$path'\n";
+}
+
+sub lock_job_state {
+    my ($jobid, $type, $sub) = @_;
+
+    my $filename = "$lock_dir/$type-$jobid.lck";
+
+    my $res = PVE::Tools::lock_file($filename, 10, $sub);
+    die $@ if $@;
+
+    return $res;
+}
+
+my $get_job_task_status = sub {
+    my ($state) = @_;
+
+    if (!defined($state->{upid})) {
+        return; # not started
+    }
+
+    my ($task, $filename) = PVE::Tools::upid_decode($state->{upid}, 1);
+    die "unable to parse worker upid - $state->{upid}\n" if !$task;
+    die "no such task\n" if !-f $filename;
+
+    my $pstart = PVE::ProcFSTools::read_proc_starttime($task->{pid});
+    if ($pstart && $pstart == $task->{pstart}) {
+        return; # still running
+    }
+
+    return PVE::Tools::upid_read_status($state->{upid});
+};
+
+# checks if the job is already finished if it was started before and
+# updates the statefile accordingly
+sub update_job_stopped {
+    my ($jobid, $type) = @_;
+
+    # first check unlocked to save time,
+    my $state = read_job_state($jobid, $type);
+    return if !defined($state) || $state->{state} ne 'started'; # removed or not started
+
+    if (defined($get_job_task_status->($state))) {
+        lock_job_state(
+            $jobid,
+            $type,
+            sub {
+                my $state = read_job_state($jobid, $type);
+                return if !defined($state) || $state->{state} ne 'started'; # removed or not started
+
+                my $new_state = {
+                    state => 'stopped',
+                    msg => $get_job_task_status->($state) // 'internal error',
+                    upid => $state->{upid},
+                    config => $state->{config},
+                };
+
+                if ($state->{updated}) { # save updated time stamp
+                    $new_state->{updated} = $state->{updated};
+                }
+
+                my $path = $get_state_file->($jobid, $type);
+                PVE::Tools::file_set_contents($path, encode_json($new_state));
+            },
+        );
+    }
+}
+
+# must be called when the job is first created
+sub create_job {
+    my ($jobid, $type, $cfg) = @_;
+
+    lock_job_state(
+        $jobid,
+        $type,
+        sub {
+            my $state = read_job_state($jobid, $type) // $default_state;
+
+            if ($state->{state} ne 'created') {
+                die "job state already exists\n";
+            }
+
+            $state->{time} = time();
+            for my $prop (@$saved_config_props) {
+                if (defined($cfg->{$prop})) {
+                    $state->{config}->{$prop} = $cfg->{$prop};
+                }
+            }
+
+            my $path = $get_state_file->($jobid, $type);
+            PVE::Tools::file_set_contents($path, encode_json($state));
+        },
+    );
+}
+
+# to be called when the job is removed
+sub remove_job {
+    my ($jobid, $type) = @_;
+    my $path = $get_state_file->($jobid, $type);
+    unlink $path;
+}
+
+# checks if the job can be started and sets the state to 'starting'
+# returns 1 if the job can be started, 0 otherwise
+sub starting_job {
+    my ($jobid, $type) = @_;
+
+    # first check unlocked to save time
+    my $state = read_job_state($jobid, $type);
+    return 0 if !defined($state) || $state->{state} eq 'started'; # removed or already started
+
+    lock_job_state(
+        $jobid,
+        $type,
+        sub {
+            my $state = read_job_state($jobid, $type);
+            return 0 if !defined($state) || $state->{state} eq 'started'; # removed or already started
+
+            my $new_state = {
+                state => 'starting',
+                time => time(),
+                config => $state->{config},
+            };
+
+            my $path = $get_state_file->($jobid, $type);
+            PVE::Tools::file_set_contents($path, encode_json($new_state));
+        },
+    );
+    return 1;
+}
+
+sub started_job {
+    my ($jobid, $type, $upid, $msg) = @_;
+
+    lock_job_state(
+        $jobid,
+        $type,
+        sub {
+            my $state = read_job_state($jobid, $type);
+            return if !defined($state); # job was removed, do not update
+            die "unexpected state '$state->{state}'\n" if $state->{state} ne 'starting';
+
+            my $new_state;
+            if (defined($msg)) {
+                $new_state = {
+                    state => 'stopped',
+                    msg => $msg,
+                    time => time(),
+                };
+            } else {
+                $new_state = {
+                    state => 'started',
+                    upid => $upid,
+                };
+            }
+            $new_state->{config} = $state->{config};
+
+            my $path = $get_state_file->($jobid, $type);
+            PVE::Tools::file_set_contents($path, encode_json($new_state));
+        },
+    );
+}
+
+# will be called when the job schedule is updated
+sub update_last_runtime {
+    my ($jobid, $type) = @_;
+    lock_job_state(
+        $jobid,
+        $type,
+        sub {
+            my $old_state = read_job_state($jobid, $type) // $default_state;
+
+            $old_state->{updated} = time();
+
+            my $path = $get_state_file->($jobid, $type);
+            PVE::Tools::file_set_contents($path, encode_json($old_state));
+        },
+    );
+}
+
+sub get_last_runtime {
+    my ($jobid, $type) = @_;
+
+    my $state = read_job_state($jobid, $type) // $default_state;
+
+    return $state->{updated} if defined($state->{updated});
+
+    if (my $upid = $state->{upid}) {
+        my ($task) = PVE::Tools::upid_decode($upid, 1);
+        die "unable to parse worker upid\n" if !$task;
+        return $task->{starttime};
+    }
+
+    return $state->{time} // 0;
+}
+
+sub run_jobs {
+    my ($first_run) = @_;
+
+    synchronize_job_states_with_config();
+
+    my $jobs_cfg = cfs_read_file('jobs.cfg');
+    my $nodename = PVE::INotify::nodename();
+
+    foreach my $id (sort keys %{ $jobs_cfg->{ids} }) {
+        my $cfg = $jobs_cfg->{ids}->{$id};
+        my $type = $cfg->{type};
+        my $schedule = delete $cfg->{schedule};
+
+        # only schedule local jobs
+        next if defined($cfg->{node}) && $cfg->{node} ne $nodename;
+
+        eval { update_job_stopped($id, $type) };
+        if (my $err = $@) {
+            warn "could not update job state, skipping - $err\n";
+            next;
+        }
+
+        # update last runtime on the first run when 'repeat-missed' is 0, so that a missed job
+        # will not start immediately after boot
+        update_last_runtime($id, $type) if $first_run && !$cfg->{'repeat-missed'};
+
+        next if defined($cfg->{enabled}) && !$cfg->{enabled}; # only schedule actually enabled jobs
+
+        my $last_run = get_last_runtime($id, $type);
+        my $calspec = PVE::CalendarEvent::parse_calendar_event($schedule);
+        my $next_sync = PVE::CalendarEvent::compute_next_event($calspec, $last_run);
+
+        next if !defined($next_sync) || time() < $next_sync; # not yet its (next) turn
+
+        my $plugin = PVE::Job::Registry->lookup($type);
+        if (starting_job($id, $type)) {
+            PVE::Cluster::cfs_update();
+
+            my $upid = eval { $plugin->run($cfg, $id, $schedule) };
+            if (my $err = $@) {
+                warn $@ if $@;
+                started_job($id, $type, undef, $err);
+            } elsif ($upid eq 'OK') { # some jobs return OK immediately
+                started_job($id, $type, undef, 'OK');
+            } else {
+                started_job($id, $type, $upid);
+            }
+        }
+    }
+}
+
+# creates and removes statefiles for job configs
+sub synchronize_job_states_with_config {
+    cfs_lock_file(
+        'jobs.cfg',
+        undef,
+        sub {
+            my $data = cfs_read_file('jobs.cfg');
+
+            for my $id (keys $data->{ids}->%*) {
+                my $job = $data->{ids}->{$id};
+                my $type = $job->{type};
+
+                my $path = $get_state_file->($id, $type);
+                if (-e $path) {
+                    detect_changed_runtime_props($id, $type, $job);
+                } else {
+                    create_job($id, $type, $job);
+                }
+            }
+
+            my $valid_types = PVE::Job::Registry->lookup_types();
+            my $type_regex = join("|", $valid_types->@*);
+
+            PVE::Tools::dir_glob_foreach(
+                $state_dir,
+                "(${type_regex})-(.*).json",
+                sub {
+                    my ($path, $type, $id) = @_;
+
+                    if (!defined($data->{ids}->{$id})) {
+                        remove_job($id, $type);
+                    }
+                },
+            );
+        },
+    );
+    die $@ if $@;
+}
+
+sub setup_dirs {
+    mkdir $state_dir;
+    mkdir $lock_dir;
+}
+
+1;