API Massive refactor of Versioned

API Implemented recursive versioned actions (publish, etc)
API Un-deprecate delete batch action
API Changed Versioned getters and setters for stages
BUG Inject query parameters during DataObject construction to prevent incorrect lazy loading

admin/code/LeftAndMain.php

@@ -521,7 +521,7 @@ class LeftAndMain extends Controller implements PermissionProvider {
 		Config::inst()->update('SSViewer', 'theme_enabled', false);
 
 		//set the reading mode for the admin to stage
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 	}
 
 	public function handleRequest(SS_HTTPRequest $request, DataModel $model = null) {

control/Director.php

@@ -268,7 +268,7 @@ class Director implements TemplateGlobalProvider {
 
 		// These are needed so that calling Director::test() does not muck with whoever is calling it.
 		// Really, it's some inappropriate coupling and should be resolved by making less use of statics.
-		$oldStage = Versioned::current_stage();
+		$oldStage = Versioned::get_stage();
 		$getVars = array();
 
 		if(!$httpMethod) $httpMethod = ($postVars || is_array($postVars)) ? "POST" : "GET";
@@ -308,7 +308,7 @@ class Director implements TemplateGlobalProvider {
 
 			// These are needed so that calling Director::test() does not muck with whoever is calling it.
 			// Really, it's some inappropriate coupling and should be resolved by making less use of statics
-			Versioned::reading_stage($oldStage);
+			Versioned::set_stage($oldStage);
 
 			Injector::unnest(); // Restore old CookieJar, etc
 			Config::unnest();

dev/DevelopmentAdmin.php

@@ -76,7 +76,7 @@ class DevelopmentAdmin extends Controller {
 		// Backwards compat: Default to "draft" stage, which is important
 		// for tasks like dev/build which call DataObject->requireDefaultRecords(),
 		// but also for other administrative tasks which have assumptions about the default stage.
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 	}
 
 	public function index() {

docs/en/02_Developer_Guides/00_Model/10_Versioning.md

@@ -13,14 +13,32 @@ Versioning in SilverStripe is handled through the [api:Versioned] class. As a [a
 be applied to any [api:DataObject] subclass. The extension class will automatically update read and write operations
 done via the ORM via the `augmentSQL` database hook.
 
-Adding Versioned to your `DataObject` subclass works the same as any other extension. It accepts two or more arguments 
-denoting the different "stages", which map to different database tables. 
+Adding Versioned to your `DataObject` subclass works the same as any other extension. It has one of two behaviours,
+which can be applied via the constructor argument.
+
+By default, adding the `Versioned extension will create a "Stage" and "Live" stage on your model, and will
+also track versioned history.
+
+
+	:::php
+	class MyStagedModel extends DataObject {
+		private staic $extensions = [
+			"Versioned"
+		];
+	}
+
+
+Alternatively, staging can be disabled, so that only versioned changes are tracked for your model. This
+can be specified by setting the constructor argument to "Versioned"
+
+
+	:::php
+	class VersionedModel extends DataObject {
+		private staic $extensions = [
+			"Versioned('Versioned')"
+		];
+	}
 
-**mysite/_config/app.yml**
-	:::yml
-	MyRecord:
-	  extensions:
-	    - Versioned("Stage","Live")
 
 <div class="notice" markdown="1">
 The extension is automatically applied to `SiteTree` class. For more information on extensions see 
@@ -34,8 +52,9 @@ of `DataObject`. Adding this extension to children of the base class will have u
 
 ## Database Structure
 
-Depending on how many stages you configured, two or more new tables will be created for your records. In the above, this
-will create a new `MyRecord_Live` table once you've rebuilt the database.
+Depending on whether staging is enabled, one or more new tables will be created for your records. `<class>_versions`
+is always created to track historic versions for your model. If staging is enabled this will also create a new
+`<class>_Live` table once you've rebuilt the database.
 
 <div class="notice" markdown="1">
 Note that the "Stage" naming has a special meaning here, it will leave the original table name unchanged, rather than 
@@ -144,6 +163,9 @@ and has_one/has_many, however it relies on a pre-existing relationship to functi
 For instance, in order to specify this dependency, you must apply `owns` and `owned_by` config
 on a relationship.
 
+When pages of type `MyPage` are published, any owned images and banners will be automatically published,
+without requiring any custom code.
+
 
 	:::php
 	class MyPage extends Page {

docs/en/04_Changelogs/4.0.0.md

@@ -7,6 +7,7 @@
  * Minimum PHP version raised to 5.5.0
  * Deprecate `SQLQuery` in favour `SQLSelect`
  * `DataList::filter` by null now internally generates "IS NULL" or "IS NOT NULL" conditions appropriately on queries
+ * `DataObject` constructor now has an additional parameter, which must be included in subclasses.
  * `DataObject::database_fields` now returns all fields on that table.
  * `DataObject::db` now returns composite fields.
  * `DataObject::ClassName` field has been refactored into a `DBClassName` type field.
@@ -34,6 +35,10 @@
    definitions of the [behat-extension](https://github.com/silverstripe-labs/silverstripe-behat-extension) module).
    Run `composer require --dev 'phpunit/phpunit:~4.8'` on existing projects to pull in the new dependency.
  * `Object::invokeWithExtensions` now has the same method signature as `Object::extend` and behaves the same way.
+ * `CMSMain` model actions have been changed:
+   * `CMSBatchAction_Delete` and `CMSMain::delete` are no longer deprecated.
+   * `CMSBatchAction_DeleteFromLive` is removed.
+   * `CMSMain.enabled_legacy_actions` config is removed.
 
 ## New API
 
@@ -69,6 +74,18 @@
    automatically check `can*()` permissions, and must be done by usercode before invocation.
  * GridField edit form now has improved support for versioned DataObjects, with basic publishing
    actions available when editing records.
+ * `Versioned` API has some breaking changes:
+   * Versioned constructor now only allows a single string to declare whether staging is enabled or not. The
+     number of names of stages are no longer able to be specified. See below for upgrading notes for models
+     with custom stages.
+   * `reading_stage` is now `set_stage`
+   * `current_stage` is now `get_stage`
+   * `getVersionedStages` is gone.
+   * `get_live_stage` is removed. Use the `Versioned::LIVE` constant instead.
+   * `getDefaultStage` is removed. Use the `Versioned::DRAFT` constant instead.
+   * `$versionableExtensions` is now `private static` instead of `protected static`
+   * `hasStages` is addded to check if an object has a given stage.
+   * `stageTable` is added to get the table for a given class and stage.
 
 ### Front-end build tooling for CMS interface
 
@@ -647,8 +664,42 @@ has been removed from core. You can configure the built-in `charmap` plugin inst
 For more information on available options and plugins please refer to the
 [tinymce documentation](https://www.tinymce.com/docs/configure/)
 
+### Upgrading DataObjects with the `Versioned` extension
+
+In most cases, versioned models with the default versioning parameters will not need to be changed. However,
+there are now additional restrictions on the use of custom stage names.
+
+Rather than declaring the list of stages a model has, the constructor for `Versioned` will take a single mode
+parameter, which declares whether or not the model is versioned and has a draft and live stage, or alternatively
+if it only has versioning without staging.
+
+For instance:
+
+
+	:::php
+	/**
+	 * This model has staging and versioning. Stages will be "Stage" and "Live"
+	 */
+	class MyStagedModel extends DataObject {
+		private staic $extensions = array(
+			"Versioned('StagedVersioned')"
+		);
+	}
+	
+	/**
+	 * This model has versioning only, and will not has a draft or live stage, nor be affected by the current stage.
+	 */
+	class MyVersionedModel extends DataObject {
+		private static $extensions = array(
+			"Versioned('Versioned')"
+		);
+	}
+
+
 ### Implementation of ownership API
 
 In order to support the recursive publishing of dataobjects, a new API has been developed to allow
 developers to declare dependencies between objects. See the
 [versioned documentation](/developer_guides/model/versioning) for more information.
+
+By default all versioned dataobjects will automatically publish objects that they own.

filesystem/AssetControlExtension.php

@@ -99,10 +99,10 @@ class AssetControlExtension extends \DataExtension
 	protected function getRecordState($record) {
 		if($this->isVersioned()) {
 			// Check stage this record belongs to
-			$stage = $record->getSourceQueryParam('Versioned.stage') ?: Versioned::current_stage();
+			$stage = $record->getSourceQueryParam('Versioned.stage') ?: Versioned::get_stage();
 
 			// Non-live stages are automatically non-public
-			if($stage !== Versioned::get_live_stage()) {
+			if($stage !== Versioned::LIVE) {
 				return AssetManipulationList::STATE_PROTECTED;
 			}
 		}
@@ -159,7 +159,7 @@ class AssetControlExtension extends \DataExtension
 		$stages = $this->owner->getVersionedStages(); // {@see Versioned::getVersionedStages}
 		foreach ($stages as $stage) {
 			// Skip current stage; These should be handled explicitly
-			if($stage === Versioned::current_stage()) {
+			if($stage === Versioned::get_stage()) {
 				continue;
 			}
 

filesystem/File.php

@@ -551,7 +551,7 @@ class File extends DataObject implements ShortcodeHandler, AssetContainer {
 	 * This method will update the File {@see DBFile} field value on success, so it must be called
 	 * before writing to the database
 	 *
-	 * @param bool True if changed
+	 * @return bool True if changed
 	 */
 	public function updateFilesystem() {
 		if(!$this->config()->update_filesystem) {
@@ -563,24 +563,22 @@ class File extends DataObject implements ShortcodeHandler, AssetContainer {
 			return false;
 		}
 
+		// Avoid moving files on live; Rely on this being done on stage prior to publish.
+		if(Versioned::get_stage() !== Versioned::DRAFT) {
+			return false;
+		}
+
 		// Check path updated record will point to
 		// If no changes necessary, skip
 		$pathBefore = $this->File->getFilename();
-		$pathAfter = $this->getFilename();
+		$pathAfter = $this->generateFilename();
 		if($pathAfter === $pathBefore) {
 			return false;
 		}
 
 		// Copy record to new location via stream
 		$stream = $this->File->getStream();
-		$result = $this->File->setFromStream($stream, $pathAfter);
-
-		// If the backend chose a new name, update the local record
-		if($result['Filename'] !== $pathAfter) {
-			// Correct saved folder to selected filename
-			$pathAfter = $result['Filename'];
-			$this->setFilename($pathAfter);
-		}
+		$this->File->setFromStream($stream, $pathAfter);
 		return true;
 	}
 
@@ -588,8 +586,10 @@ class File extends DataObject implements ShortcodeHandler, AssetContainer {
 	 * Collate selected descendants of this page.
 	 * $condition will be evaluated on each descendant, and if it is succeeds, that item will be added
 	 * to the $collator array.
-	 * @param condition The PHP condition to be evaluated.  The page will be called $item
-	 * @param collator An array, passed by reference, to collect all of the matching descendants.
+	 *
+	 * @param string $condition The PHP condition to be evaluated.  The page will be called $item
+	 * @param array $collator An array, passed by reference, to collect all of the matching descendants.
+	 * @return true|null
 	 */
 	public function collateDescendants($condition, &$collator) {
 		if($children = $this->Children()) {
@@ -610,7 +610,8 @@ class File extends DataObject implements ShortcodeHandler, AssetContainer {
 	 *
 	 * Does not change the filesystem itself, please use {@link write()} for this.
 	 *
-	 * @param String $name
+	 * @param string $name
+	 * @return $this
 	 */
 	public function setName($name) {
 		$oldName = $this->Name;
@@ -655,7 +656,7 @@ class File extends DataObject implements ShortcodeHandler, AssetContainer {
 			$this->Title = str_replace(array('-','_'),' ', preg_replace('/\.[^.]+$/', '', $name));
 		}
 
-		return $name;
+		return $this;
 	}
 
 	/**
@@ -704,13 +705,18 @@ class File extends DataObject implements ShortcodeHandler, AssetContainer {
 		return Director::absoluteBaseURL()."admin/assets/removefile/".$this->ID;
 	}
 
-	public function getFilename() {
+	/**
+	 * Get expected value of Filename tuple value. Will be used to trigger
+	 * a file move on draft stage.
+	 *
+	 * @return string
+	 */
+	public function generateFilename() {
 		// Check if this file is nested within a folder
 		$parent = $this->Parent();
 		if($parent && $parent->exists()) {
 			return $this->join_paths($parent->getFilename(), $this->Name);
 		}
-
 		return $this->Name;
 	}
 
@@ -1027,6 +1033,10 @@ class File extends DataObject implements ShortcodeHandler, AssetContainer {
 		return false;
 	}
 
+	public function getFilename() {
+		return $this->File->Filename;
+	}
+
 	public function getHash() {
 		return $this->File->Hash;
 	}

filesystem/FileMigrationHelper.php

@@ -34,8 +34,8 @@ class FileMigrationHelper extends Object {
 
 		// Loop over all files
 		$count = 0;
-		$originalState = \Versioned::get_reading_mode();
-		\Versioned::reading_stage('Stage');
+		$originalState = Versioned::get_reading_mode();
+		Versioned::set_stage(Versioned::DRAFT);
 		$filenameMap = $this->getFilenameArray();
 		foreach($this->getFileQuery() as $file) {
 			// Get the name of the file to import
@@ -45,7 +45,7 @@ class FileMigrationHelper extends Object {
 				$count++;
 			}
 		}
-		\Versioned::set_reading_mode($originalState);
+		Versioned::set_reading_mode($originalState);
 		return $count;
 	}
 
@@ -54,7 +54,7 @@ class FileMigrationHelper extends Object {
 	 *
 	 * @param string $base Absolute base path (parent of assets folder)
 	 * @param File $file
-	 * @param type $legacyFilename
+	 * @param string $legacyFilename
 	 * @return bool True if this file is imported successfully
 	 */
 	protected function migrateFile($base, File $file, $legacyFilename) {
@@ -64,8 +64,9 @@ class FileMigrationHelper extends Object {
 			return false;
 		}
 
+
 		// Copy local file into this filesystem
-		$filename = $file->getFilename();
+		$filename = $file->generateFilename();
 		$result = $file->setFromLocalFile(
 			$path, $filename, null, null,
 			array('conflict' => AssetStore::CONFLICT_OVERWRITE)
@@ -73,7 +74,7 @@ class FileMigrationHelper extends Object {
 
 		// Move file if the APL changes filename value
 		if($result['Filename'] !== $filename) {
-			$this->setFilename($result['Filename']);
+			$file->setFilename($result['Filename']);
 		}
 
 		// Save and publish

filesystem/Folder.php

@@ -233,7 +233,7 @@ class Folder extends File {
 	}
 
 	public function getFilename() {
-		return parent::getFilename() . '/';
+		return parent::generateFilename() . '/';
 	}
 
 	/**
@@ -258,11 +258,24 @@ class Folder extends File {
 	public function onAfterWrite() {
 		parent::onAfterWrite();
 
-		// Ensure that children loading $this->Parent() load the refreshed record
-		$this->flushCache();
+		// No publishing UX for folders, so just cascade changes live
+		if(Versioned::get_stage() === Versioned::DRAFT) {
+			$this->publish(Versioned::DRAFT, Versioned::LIVE);
+		}
+
+		// Update draft version of all child records
 		$this->updateChildFilesystem();
 	}
 
+	public function onAfterDelete() {
+		parent::onAfterDelete();
+
+		// Cascade deletions to live
+		if(Versioned::get_stage() === Versioned::DRAFT) {
+			$this->deleteFromStage(Versioned::LIVE);
+		}
+	}
+
 	public function updateFilesystem() {
 		// No filesystem changes to update
 	}
@@ -278,6 +291,12 @@ class Folder extends File {
 	 * Update filesystem of all children
 	 */
 	public function updateChildFilesystem() {
+		// Don't synchronise on live (rely on publishing instead)
+		if(Versioned::get_stage() === Versioned::LIVE) {
+			return;
+		}
+
+		$this->flushCache();
 		// Writing this record should trigger a write (and potential updateFilesystem) on each child
 		foreach ($this->AllChildren() as $child) {
 			$child->write();

model/DataList.php

@@ -748,10 +748,8 @@ class DataList extends ViewableData implements SS_List, SS_Filterable, SS_Sortab
 		if(class_exists($row['RecordClassName'])) {
 			$class = $row['RecordClassName'];
 		}
-		$item = Injector::inst()->create($class, $row, false, $this->model);
 
-		//set query params on the DataObject to tell the lazy loading mechanism the context the object creation context
-		$item->setSourceQueryParams($this->getQueryParams());
+		$item = Injector::inst()->create($class, $row, false, $this->model, $this->getQueryParams());
 
 		return $item;
 	}

model/DataObject.php

@@ -398,10 +398,15 @@ class DataObject extends ViewableData implements DataObjectInterface, i18nEntity
 	 * field values.  Normally this contructor is only used by the internal systems that get objects from the database.
 	 * @param boolean $isSingleton This this to true if this is a singleton() object, a stub for calling methods.
 	 *                             Singletons don't have their defaults set.
+	 * @param DataModel $model
+	 * @param array $queryParams List of DataQuery params necessary to lazy load, or load related objects.
 	 */
-	public function __construct($record = null, $isSingleton = false, $model = null) {
+	public function __construct($record = null, $isSingleton = false, $model = null, $queryParams = array()) {
 		parent::__construct();
 
+		// Set query params on the DataObject to tell the lazy loading mechanism the context the object creation context
+		$this->setSourceQueryParams($queryParams);
+
 		// Set the fields data.
 		if(!$record) {
 			$record = array(
@@ -2313,7 +2318,7 @@ class DataObject extends ViewableData implements DataObjectInterface, i18nEntity
 	/**
 	 * Loads all the stub fields that an initial lazy load didn't load fully.
 	 *
-	 * @param tableClass Base table to load the values from. Others are joined as required.
+	 * @param string $tableClass Base table to load the values from. Others are joined as required.
 	 * Not specifying a tableClass will load all lazy fields from all tables.
 	 */
 	protected function loadLazyFields($tableClass = null) {
@@ -2334,17 +2339,22 @@ class DataObject extends ViewableData implements DataObjectInterface, i18nEntity
 
 		// Reset query parameter context to that of this DataObject
 		if($params = $this->getSourceQueryParams()) {
-			foreach($params as $key => $value) $dataQuery->setQueryParam($key, $value);
+			foreach($params as $key => $value) {
+				$dataQuery->setQueryParam($key, $value);
+			}
 		}
 
 		// TableField sets the record ID to "new" on new row data, so don't try doing anything in that case
-		if(!is_numeric($this->record['ID'])) return false;
+		if(!is_numeric($this->record['ID'])) {
+			return;
+		}
 
 		// Limit query to the current record, unless it has the Versioned extension,
 		// in which case it requires special handling through augmentLoadLazyFields()
-		if(!$this->hasExtension('Versioned')) {
-			$dataQuery->where("\"$tableClass\".\"ID\" = {$this->record['ID']}")->limit(1);
-		}
+		$baseTable = ClassInfo::baseDataClass($this);
+		$dataQuery->where([
+			"\"{$baseTable}\".\"ID\"" => $this->record['ID']
+		])->limit(1);
 
 		$columns = array();
 

model/Image.php

@@ -7,8 +7,8 @@
  * @subpackage filesystem
  */
 class Image extends File {
-	public function __construct($record = null, $isSingleton = false, $model = null) {
-		parent::__construct($record, $isSingleton, $model);
+	public function __construct($record = null, $isSingleton = false, $model = null, $queryParams = array()) {
+		parent::__construct($record, $isSingleton, $model, $queryParams);
 		$this->File->setAllowedCategories('image/supported');
 	}
 

model/RelationList.php

@@ -22,7 +22,7 @@ abstract class RelationList extends DataList implements Relation {
 		// Remove `Foreign.` query parameters for created objects,
 		// as this would interfere with relations on those objects.
 		foreach(array_keys($params) as $key) {
-			if(stripos($key, 'Foreign.') !== 0) {
+			if(stripos($key, 'Foreign.') === 0) {
 				unset($params[$key]);
 			}
 		}

model/fieldtypes/Datetime.php

@@ -177,6 +177,7 @@ class SS_Datetime extends Date implements TemplateGlobalProvider {
 	 * Caution: This sets a fixed date that doesn't increment with time.
 	 *
 	 * @param SS_Datetime|string $datetime Either in object format, or as a SS_Datetime compatible string.
+	 * @throws Exception
 	 */
 	public static function set_mock_now($datetime) {
 		if($datetime instanceof SS_Datetime) {

model/queries/SQLSelect.php

@@ -61,7 +61,7 @@ class SQLSelect extends SQLConditionalExpression {
 	/**
 	 * Construct a new SQLSelect.
 	 *
-	 * @param array $select An array of SELECT fields.
+	 * @param array|string $select An array of SELECT fields.
 	 * @param array|string $from An array of FROM clauses. The first one should be just the table name.
 	 * Each should be ANSI quoted.
 	 * @param array $where An array of WHERE clauses.

model/versioning/Versioned.php

@@ -14,29 +14,42 @@
  * @subpackage model
  */
 class Versioned extends DataExtension implements TemplateGlobalProvider {
-	/**
-	 * An array of possible stages.
-	 * @var array
-	 */
-	protected $stages;
 
 	/**
-	 * The 'default' stage.
+	 * Versioning mode for this object.
+	 * Note: Not related to the current versioning mode in the state / session
+	 * Will be one of 'StagedVersioned' or 'Versioned';
+	 *
 	 * @var string
 	 */
-	protected $defaultStage;
-
-	/**
-	 * The 'live' stage.
-	 * @var string
-	 */
-	protected $liveStage;
+	protected $mode;
 
 	/**
 	 * The default reading mode
 	 */
 	const DEFAULT_MODE = 'Stage.Live';
 
+	/**
+	 * Constructor arg to specify that staging is active on this record.
+	 * 'Staging' implies that 'Versioning' is also enabled.
+	 */
+	const STAGEDVERSIONED = 'StagedVersioned';
+
+	/**
+	 * Constructor arg to specify that versioning only is active on this record.
+	 */
+	const VERSIONED = 'Versioned';
+
+	/**
+	 * The Public stage.
+	 */
+	const LIVE = 'Live';
+
+	/**
+	 * The draft (default) stage
+	 */
+	const DRAFT = 'Stage';
+
 	/**
 	 * A version that a DataObject should be when it is 'migrating',
 	 * that is, when it is in the process of moving from one stage to another.
@@ -53,6 +66,8 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	protected static $cache_versionnumber;
 
 	/**
+	 * Current reading mode
+	 *
 	 * @var string
 	 */
 	protected static $reading_mode = null;
@@ -92,6 +107,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * Used to enable or disable the prepopulation of the version number cache.
 	 * Defaults to true.
 	 *
+	 * @config
 	 * @var boolean
 	 */
 	private static $prepopulate_versionnumber_cache = true;
@@ -99,6 +115,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	/**
 	 * Keep track of the archive tables that have been created.
 	 *
+	 * @config
 	 * @var array
 	 */
 	private static $archive_tables = array();
@@ -149,9 +166,10 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * Make sure your extension has a static $enabled-property that determines if it is
 	 * processed by Versioned.
 	 *
+	 * @config
 	 * @var array
 	 */
-	protected static $versionableExtensions = array('Translatable' => 'lang');
+	private static $versionableExtensions = array('Translatable' => 'lang');
 
 	/**
 	 * Permissions necessary to view records outside of the live stage (e.g. archive / draft stage).
@@ -194,29 +212,9 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public static function reset() {
 		self::$reading_mode = '';
-
 		Session::clear('readingMode');
 	}
 
-	/**
-	 * Construct a new Versioned object.
-	 *
-	 * @var array $stages The different stages the versioned object can be.
-	 * The first stage is considered the 'default' stage, the last stage is
-	 * considered the 'live' stage.
-	 */
-	public function __construct($stages = array('Stage','Live')) {
-		parent::__construct();
-
-		if(!is_array($stages)) {
-			$stages = func_get_args();
-		}
-
-		$this->stages = $stages;
-		$this->defaultStage = reset($stages);
-		$this->liveStage = array_pop($stages);
-	}
-
 	/**
 	 * Amend freshly created DataQuery objects with versioned-specific
 	 * information.
@@ -230,24 +228,119 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		if($parts[0] == 'Archive') {
 			$dataQuery->setQueryParam('Versioned.mode', 'archive');
 			$dataQuery->setQueryParam('Versioned.date', $parts[1]);
-		} else if($parts[0] == 'Stage' && in_array($parts[1], $this->stages)) {
+		} else if($parts[0] == 'Stage' && $this->hasStages()) {
 			$dataQuery->setQueryParam('Versioned.mode', 'stage');
 			$dataQuery->setQueryParam('Versioned.stage', $parts[1]);
 		}
 	}
 
+	/**
+	 * Construct a new Versioned object.
+	 *
+	 * @var string $mode One of "StagedVersioned" or "Versioned".
+	 */
+	public function __construct($mode = self::STAGEDVERSIONED) {
+		parent::__construct();
+
+		// Handle deprecated behaviour
+		if($mode === 'Stage' && func_num_args() === 1) {
+			Deprecation::notice("5.0", "Versioned now takes a mode as a single parameter");
+			$mode = static::VERSIONED;
+		} elseif(is_array($mode) || func_num_args() > 1) {
+			Deprecation::notice("5.0", "Versioned now takes a mode as a single parameter");
+			$mode = func_num_args() > 1 || count($mode) > 1
+				? static::STAGEDVERSIONED
+				: static::VERSIONED;
+		}
+
+		if(!in_array($mode, array(static::STAGEDVERSIONED, static::VERSIONED))) {
+			throw new InvalidArgumentException("Invalid mode: {$mode}");
+		}
+
+		$this->mode = $mode;
+	}
+
+	/**
+	 * Cache of version to modified dates for this objects
+	 *
+	 * @var array
+	 */
+	protected $versionModifiedCache = array();
+
+	/**
+	 * Get modified date for the given version
+	 *
+	 * @param int $version
+	 * @return string
+	 */
+	protected function getLastEditedForVersion($version) {
+		// Cache key
+		$baseTable = ClassInfo::baseDataClass($this->owner);
+		$id = $this->owner->ID;
+		$key = "{$baseTable}#{$id}/{$version}";
+
+		// Check cache
+		if(isset($this->versionModifiedCache[$key])) {
+			return $this->versionModifiedCache[$key];
+		}
+
+		// Build query
+		$table = "\"{$baseTable}_versions\"";
+		$query = SQLSelect::create('"LastEdited"', $table)
+			->addWhere([
+				"{$table}.\"RecordID\"" => $id,
+				"{$table}.\"Version\"" => $version
+			]);
+		$date = $query->execute()->value();
+		if($date) {
+			$this->versionModifiedCache[$key] = $date;
+		}
+		return $date;
+	}
+
 
 	public function updateInheritableQueryParams(&$params) {
+		// Skip if versioned isn't set
+		if(!isset($params['Versioned.mode'])) {
+			return;
+		}
+
+		// Adjust query based on original selection criterea
+		$owner = $this->owner;
+		switch($params['Versioned.mode']) {
+			case 'all_versions': {
 				// Versioned.mode === all_versions doesn't inherit very well, so default to stage
-		if(isset($params['Versioned.mode']) && $params['Versioned.mode'] === 'all_versions') {
 				$params['Versioned.mode'] = 'stage';
-			$params['Versioned.stage'] = $this->defaultStage;
+				$params['Versioned.stage'] = static::DRAFT;
+				break;
+			}
+			case 'version': {
+				// If we selected this object from a specific version, we need
+				// to find the date this version was published, and ensure
+				// inherited queries select from that date.
+				$version = $params['Versioned.version'];
+				$date = $this->getLastEditedForVersion($version);
+
+				// Filter related objects at the same date as this version
+				unset($params['Versioned.version']);
+				if($date) {
+					$params['Versioned.mode'] = 'archive';
+					$params['Versioned.date'] = $date;
+				} else {
+					// Fallback to default
+					$params['Versioned.mode'] = 'stage';
+					$params['Versioned.stage'] = static::DRAFT;
+				}
+				break;
+			}
 		}
 	}
 
 	/**
 	 * Augment the the SQLSelect that is created by the DataQuery
 	 *
+	 * See {@see augmentLazyLoadFields} for lazy-loading applied prior to this.
+	 *
 	 * @param SQLSelect $query
 	 * @param DataQuery $dataQuery
 	 * @throws InvalidArgumentException
@@ -259,62 +352,32 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 
 		$baseTable = ClassInfo::baseDataClass($dataQuery->dataClass());
 
-		switch($dataQuery->getQueryParam('Versioned.mode')) {
-		// Reading a specific data from the archive
-		case 'archive':
-			$date = $dataQuery->getQueryParam('Versioned.date');
-			foreach($query->getFrom() as $table => $dummy) {
-				if(!$this->isTableVersioned($table)) {
-					continue;
-				}
-
-				$query->renameTable($table, $table . '_versions');
-				$query->replaceText("\"{$table}_versions\".\"ID\"", "\"{$table}_versions\".\"RecordID\"");
-				$query->replaceText("`{$table}_versions`.`ID`", "`{$table}_versions`.`RecordID`");
-
-				// Add all <basetable>_versions columns
-				foreach(Config::inst()->get('Versioned', 'db_for_versions_table') as $name => $type) {
-					$query->selectField(sprintf('"%s_versions"."%s"', $baseTable, $name), $name);
-				}
-				$query->selectField(sprintf('"%s_versions"."%s"', $baseTable, 'RecordID'), "ID");
-
-				if($table != $baseTable) {
-					$query->addWhere("\"{$table}_versions\".\"Version\" = \"{$baseTable}_versions\".\"Version\"");
-				}
-			}
-			// Link to the version archived on that date
-			$query->addWhere(array(
-				"\"{$baseTable}_versions\".\"Version\" IN
-				(SELECT LatestVersion FROM
-					(SELECT
-						\"{$baseTable}_versions\".\"RecordID\",
-						MAX(\"{$baseTable}_versions\".\"Version\") AS LatestVersion
-						FROM \"{$baseTable}_versions\"
-						WHERE \"{$baseTable}_versions\".\"LastEdited\" <= ?
-						GROUP BY \"{$baseTable}_versions\".\"RecordID\"
-					) AS \"{$baseTable}_versions_latest\"
-					WHERE \"{$baseTable}_versions_latest\".\"RecordID\" = \"{$baseTable}_versions\".\"RecordID\"
-				)" => $date
-			));
-			break;
-
+		$versionedMode = $dataQuery->getQueryParam('Versioned.mode');
+		switch($versionedMode) {
 		// Reading a specific stage (Stage or Live)
 		case 'stage':
+			// Check if we need to rewrite this table
 			$stage = $dataQuery->getQueryParam('Versioned.stage');
-			if($stage && ($stage != $this->defaultStage)) {
+			if(!$this->hasStages() || $stage === static::DRAFT) {
+				break;
+			}
+			// Rewrite all tables to select from the live version
 			foreach($query->getFrom() as $table => $dummy) {
 				if(!$this->isTableVersioned($table)) {
 					continue;
 				}
-					$query->renameTable($table, $table . '_' . $stage);
-				}
+				$stageTable = $this->stageTable($table, $stage);
+				$query->renameTable($table, $stageTable);
 			}
 			break;
 
 		// Reading a specific stage, but only return items that aren't in any other stage
 		case 'stage_unique':
-			$stage = $dataQuery->getQueryParam('Versioned.stage');
+			if(!$this->hasStages()) {
+				break;
+			}
 
+			$stage = $dataQuery->getQueryParam('Versioned.stage');
 			// Recurse to do the default stage behavior (must be first, we rely on stage renaming happening before
 			// below)
 			$dataQuery->setQueryParam('Versioned.mode', 'stage');
@@ -323,11 +386,13 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 
 			// Now exclude any ID from any other stage. Note that we double rename to avoid the regular stage rename
 			// renaming all subquery references to be Versioned.stage
-			foreach($this->stages as $excluding) {
-				if ($excluding == $stage) continue;
+			foreach([static::DRAFT, static::LIVE] as $excluding) {
+				if ($excluding == $stage) {
+					continue;
+				}
 
 				$tempName = 'ExclusionarySource_'.$excluding;
-				$excludingTable = $baseTable . ($excluding && $excluding != $this->defaultStage ? "_$excluding" : '');
+				$excludingTable = $baseTable . ($excluding && $excluding != static::DRAFT ? "_$excluding" : '');
 
 				$query->addWhere('"'.$baseTable.'"."ID" NOT IN (SELECT "ID" FROM "'.$tempName.'")');
 				$query->renameTable($tempName, $excludingTable);
@@ -335,8 +400,10 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 			break;
 
 		// Return all version instances
+		case 'archive':
 		case 'all_versions':
 		case 'latest_versions':
+		case 'version':
 			foreach($query->getFrom() as $alias => $join) {
 				if(!$this->isTableVersioned($alias)) {
 					continue;
@@ -368,10 +435,32 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 				"count(DISTINCT \"{$baseTable}_versions\".\"ID\")"
 			);
 
-			// latest_version has one more step
+			// Add additional versioning filters
+			switch($versionedMode) {
+				case 'archive': {
+					$date = $dataQuery->getQueryParam('Versioned.date');
+					if(!$date) {
+						throw new InvalidArgumentException("Invalid archive date");
+					}
+					// Link to the version archived on that date
+					$query->addWhere([
+						"\"{$baseTable}_versions\".\"Version\" IN
+						(SELECT LatestVersion FROM
+							(SELECT
+								\"{$baseTable}_versions\".\"RecordID\",
+								MAX(\"{$baseTable}_versions\".\"Version\") AS LatestVersion
+								FROM \"{$baseTable}_versions\"
+								WHERE \"{$baseTable}_versions\".\"LastEdited\" <= ?
+								GROUP BY \"{$baseTable}_versions\".\"RecordID\"
+							) AS \"{$baseTable}_versions_latest\"
+							WHERE \"{$baseTable}_versions_latest\".\"RecordID\" = \"{$baseTable}_versions\".\"RecordID\"
+						)" => $date
+					]);
+					break;
+				}
+				case 'latest_versions': {
 					// Return latest version instances, regardless of whether they are on a particular stage
 					// This provides "show all, including deleted" functonality
-			if($dataQuery->getQueryParam('Versioned.mode') == 'latest_versions') {
 					$query->addWhere(
 						"\"{$baseTable}_versions\".\"Version\" IN
 						(SELECT LatestVersion FROM
@@ -382,10 +471,26 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 								GROUP BY \"{$baseTable}_versions\".\"RecordID\"
 							) AS \"{$baseTable}_versions_latest\"
 							WHERE \"{$baseTable}_versions_latest\".\"RecordID\" = \"{$baseTable}_versions\".\"RecordID\"
-					)");
-			} else {
+						)"
+					);
+					break;
+				}
+				case 'version': {
+					// If selecting a specific version, filter it here
+					$version = $dataQuery->getQueryParam('Versioned.version');
+					if(!$version) {
+						throw new InvalidArgumentException("Invalid version");
+					}
+					$query->addWhere([
+						"\"{$baseTable}_versions\".\"Version\"" => $version
+					]);
+					break;
+				}
+				default: {
 					// If all versions are requested, ensure that records are sorted by this field
 					$query->addOrderBy(sprintf('"%s_versions"."%s"', $baseTable, 'Version'));
+					break;
+				}
 			}
 			break;
 		default:
@@ -422,18 +527,18 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		// metadata set on the query object. This prevents regular queries from
 		// accidentally querying the *_versions tables.
 		$versionedMode = $dataObject->getSourceQueryParam('Versioned.mode');
-		$dataClass = $dataQuery->dataClass();
-		$modesToAllowVersioning = array('all_versions', 'latest_versions', 'archive');
+		$dataClass = ClassInfo::baseDataClass($dataQuery->dataClass());
+		$modesToAllowVersioning = array('all_versions', 'latest_versions', 'archive', 'version');
 		if(
 			!empty($dataObject->Version) &&
 			(!empty($versionedMode) && in_array($versionedMode,$modesToAllowVersioning))
 		) {
-			$dataQuery->where("\"$dataClass\".\"RecordID\" = " . $dataObject->ID);
-			$dataQuery->where("\"$dataClass\".\"Version\" = " . $dataObject->Version);
+			// This will ensure that augmentSQL will select only the same version as the owner,
+			// regardless of how this object was initially selected
+			$dataQuery->where([
+				"\"$dataClass\".\"Version\"" => $dataObject->Version
+			]);
 			$dataQuery->setQueryParam('Versioned.mode', 'all_versions');
-		} else {
-			// Same behaviour as in DataObject->loadLazyFields
-			$dataQuery->where("\"$dataClass\".\"ID\" = {$dataObject->ID}")->limit(1);
 		}
 	}
 
@@ -446,26 +551,27 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	public static function on_db_reset() {
 		// Drop all temporary tables
 		$db = DB::get_conn();
-		foreach(self::$archive_tables as $tableName) {
+		foreach(static::$archive_tables as $tableName) {
 			if(method_exists($db, 'dropTable')) $db->dropTable($tableName);
 			else $db->query("DROP TABLE \"$tableName\"");
 		}
 
 		// Remove references to them
-		self::$archive_tables = array();
+		static::$archive_tables = array();
 	}
 
 	public function augmentDatabase() {
-		$classTable = $this->owner->class;
+		$owner = $this->owner;
+		$classTable = $owner->class;
 
-		$isRootClass = ($this->owner->class == ClassInfo::baseDataClass($this->owner->class));
+		$isRootClass = ($owner->class == ClassInfo::baseDataClass($owner->class));
 
 		// Build a list of suffixes whose tables need versioning
 		$allSuffixes = array();
-		$versionableExtensions = $this->owner->config()->versionableExtensions;
+		$versionableExtensions = $owner->config()->versionableExtensions;
 		if(count($versionableExtensions)){
 			foreach ($versionableExtensions as $versionableExtension => $suffixes) {
-			if ($this->owner->hasExtension($versionableExtension)) {
+				if ($owner->hasExtension($versionableExtension)) {
 					$allSuffixes = array_merge($allSuffixes, (array)$suffixes);
 					foreach ((array)$suffixes as $suffix) {
 						$allSuffixes[$suffix] = $versionableExtension;
@@ -484,14 +590,14 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 			if ($suffix) $table = "{$classTable}_$suffix";
 			else $table = $classTable;
 
-			$fields = DataObject::database_fields($this->owner->class);
+			$fields = DataObject::database_fields($owner->class);
 			unset($fields['ID']);
 			if($fields) {
-				$options = Config::inst()->get($this->owner->class, 'create_table_options', Config::FIRST_SET);
-				$indexes = $this->owner->databaseIndexes();
-				if ($suffix && ($ext = $this->owner->getExtensionInstance($allSuffixes[$suffix]))) {
+				$options = Config::inst()->get($owner->class, 'create_table_options', Config::FIRST_SET);
+				$indexes = $owner->databaseIndexes();
+				if ($suffix && ($ext = $owner->getExtensionInstance($allSuffixes[$suffix]))) {
 					if (!$ext->isVersionedTable($table)) continue;
-					$ext->setOwner($this->owner);
+					$ext->setOwner($owner);
 					$fields = $ext->fieldsInExtraTables($suffix);
 					$ext->clearOwner();
 					$indexes = $fields['indexes'];
@@ -499,24 +605,13 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 				}
 
 				// Create tables for other stages
-				foreach($this->stages as $stage) {
+				if($this->hasStages()) {
 					// Extra tables for _Live, etc.
 					// Change unique indexes to 'index'.  Versioned tables may run into unique indexing difficulties
 					// otherwise.
+					$liveTable = $this->stageTable($table, static::LIVE);
 					$indexes = $this->uniqueToIndex($indexes);
-					if($stage != $this->defaultStage) {
-						DB::require_table("{$table}_$stage", $fields, $indexes, false, $options);
-					}
-
-					// Version fields on each root table (including Stage)
-					/*
-					if($isRootClass) {
-						$stageTable = ($stage == $this->defaultStage) ? $table : "{$table}_$stage";
-						$parts=Array('datatype'=>'int', 'precision'=>11, 'null'=>'not null', 'default'=>(int)0);
-						$values=Array('type'=>'int', 'parts'=>$parts);
-						DB::requireField($stageTable, 'Version', $values);
-					}
-					*/
+					DB::require_table($liveTable, $fields, $indexes, false, $options);
 				}
 
 				if($isRootClass) {
@@ -611,8 +706,9 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 				DB::require_table("{$table}_versions", $versionFields, $versionIndexes, true, $options);
 			} else {
 				DB::dont_require_table("{$table}_versions");
-				foreach($this->stages as $stage) {
-					if($stage != $this->defaultStage) DB::dont_require_table("{$table}_$stage");
+				if($this->hasStages()) {
+					$liveTable = $this->stageTable($table, static::LIVE);
+					DB::dont_require_table($liveTable);
 				}
 			}
 		}
@@ -724,7 +820,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 			);
 		}
 
-		$newTable = $table . '_' . Versioned::current_stage();
+		$newTable = $this->stageTable($table, Versioned::get_stage());
 		$manipulation[$newTable] = $manipulation[$table];
 		unset($manipulation[$table]);
 	}
@@ -733,7 +829,8 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	public function augmentWrite(&$manipulation) {
 		// get Version number from base data table on write
 		$version = null;
-		$baseDataClass = ClassInfo::baseDataClass($this->owner->class);
+		$owner = $this->owner;
+		$baseDataClass = ClassInfo::baseDataClass($owner->class);
 		if(isset($manipulation[$baseDataClass]['fields'])) {
 			if ($this->migratingVersion) {
 				$manipulation[$baseDataClass]['fields']['Version'] = $this->migratingVersion;
@@ -781,11 +878,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 			}
 
 			// If we're editing Live, then use (table)_Live instead of (table)
-			if(
-				Versioned::current_stage()
-				&& Versioned::current_stage() != $this->defaultStage
-				&& in_array(Versioned::current_stage(), $this->stages)
-			) {
+			if($this->hasStages() && static::get_stage() === static::LIVE) {
 				$this->augmentWriteStaged($manipulation, $table, $id);
 			}
 		}
@@ -798,7 +891,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		// Add the new version # back into the data object, for accessing
 		// after this write
 		if(isset($thisVersion)) {
-			$this->owner->Version = str_replace("'","", $thisVersion);
+			$owner->Version = str_replace("'","", $thisVersion);
 		}
 	}
 
@@ -872,25 +965,26 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		}
 
 		// Skip search for unsaved records
-		if(!$this->owner->isInDB()) {
+		$owner = $this->owner;
+		if(!$owner->isInDB()) {
 			return $list;
 		}
 
-		$relationships = $this->owner->config()->{$source};
+		$relationships = $owner->config()->{$source};
 		foreach($relationships as $relationship) {
 			// Warn if invalid config
-			if(!$this->owner->hasMethod($relationship)) {
+			if(!$owner->hasMethod($relationship)) {
 				trigger_error(sprintf(
 					"Invalid %s config value \"%s\" on object on class \"%s\"",
 					$source,
 					$relationship,
-					$this->owner->class
+					$owner->class
 				), E_USER_WARNING);
 				continue;
 			}
 
 			// Inspect value of this relationship
-			$items = $this->owner->{$relationship}();
+			$items = $owner->{$relationship}();
 			if(!$items) {
 				continue;
 			}
@@ -944,13 +1038,14 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		}
 
 		// Standard mechanism for accepting permission changes from extensions
-		$extended = $this->owner->extendedCan('canPublish', $member);
+		$owner = $this->owner;
+		$extended = $owner->extendedCan('canPublish', $member);
 		if($extended !== null) {
 			return $extended;
 		}
 
 		// Default to relying on edit permission
-		return $this->owner->canEdit($member);
+		return $owner->canEdit($member);
 	}
 
 	/**
@@ -974,13 +1069,14 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		}
 
 		// Standard mechanism for accepting permission changes from extensions
-		$extended = $this->owner->extendedCan('canUnpublish', $member);
+		$owner = $this->owner;
+		$extended = $owner->extendedCan('canUnpublish', $member);
 		if($extended !== null) {
 			return $extended;
 		}
 
 		// Default to relying on canPublish
-		return $this->owner->canPublish($member);
+		return $owner->canPublish($member);
 	}
 
 	/**
@@ -1005,24 +1101,62 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		}
 
 		// Standard mechanism for accepting permission changes from extensions
-		$extended = $this->owner->extendedCan('canArchive', $member);
+		$owner = $this->owner;
+		$extended = $owner->extendedCan('canArchive', $member);
 		if($extended !== null) {
             return $extended;
         }
 
 		// Check if this record can be deleted from stage
-        if(!$this->owner->canDelete($member)) {
+        if(!$owner->canDelete($member)) {
             return false;
         }
 
         // Check if we can delete from live
-        if(!$this->owner->canUnpublish($member)) {
+        if(!$owner->canUnpublish($member)) {
             return false;
         }
 
 		return true;
 	}
 
+	/**
+	 * Check if the user can revert this record to live
+	 *
+	 * @param Member $member
+	 * @return bool
+	 */
+	public function canRevertToLive($member = null) {
+		$owner = $this->owner;
+
+		// Skip if invoked by extendedCan()
+		if(func_num_args() > 4) {
+			return null;
+		}
+
+		// Can't revert if not on live
+		if(!$owner->isPublished()) {
+			return false;
+		}
+
+		if(!$member) {
+            $member = Member::currentUser();
+        }
+
+		if(Permission::checkMember($member, "ADMIN")) {
+			return true;
+		}
+
+		// Standard mechanism for accepting permission changes from extensions
+		$extended = $owner->extendedCan('canRevertToLive', $member);
+		if($extended !== null) {
+            return $extended;
+        }
+
+		// Default to canEdit
+		return $owner->canEdit($member);
+	}
+
 	/**
 	 * Extend permissions to include additional security for objects that are not published to live.
 	 *
@@ -1054,9 +1188,10 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public function canViewVersioned($member = null) {
 		// Bypass when live stage
-		$mode = $this->owner->getSourceQueryParam("Versioned.mode");
-		$stage = $this->owner->getSourceQueryParam("Versioned.stage");
-		if ($mode === 'stage' && $stage === static::get_live_stage()) {
+		$owner = $this->owner;
+		$mode = $owner->getSourceQueryParam("Versioned.mode");
+		$stage = $owner->getSourceQueryParam("Versioned.stage");
+		if ($mode === 'stage' && $stage === static::LIVE) {
 			return true;
 		}
 
@@ -1066,26 +1201,26 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		}
 
 		// Bypass if record doesn't have a live stage
-		if(!in_array(static::get_live_stage(), $this->getVersionedStages())) {
+		if(!$this->hasStages()) {
 			return true;
 		}
 
 		// If we weren't definitely loaded from live, and we can't view non-live content, we need to
 		// check to make sure this version is the live version and so can be viewed
-		$latestVersion = Versioned::get_versionnumber_by_stage($this->owner->class, 'Live', $this->owner->ID);
-		if ($latestVersion == $this->owner->Version) {
+		$latestVersion = Versioned::get_versionnumber_by_stage($owner->class, static::LIVE, $owner->ID);
+		if ($latestVersion == $owner->Version) {
 			// Even if this is loaded from a non-live stage, this is the live version
 			return true;
 		}
 
 		// Extend versioned behaviour
-		$extended = $this->owner->extendedCan('canViewNonLive', $member);
+		$extended = $owner->extendedCan('canViewNonLive', $member);
 		if($extended !== null) {
 			return (bool)$extended;
 		}
 
 		// Fall back to default permission check
-		$permissions = Config::inst()->get($this->owner->class, 'non_live_permissions', Config::FIRST_SET);
+		$permissions = Config::inst()->get($owner->class, 'non_live_permissions', Config::FIRST_SET);
 		$check = Permission::checkMember($member, $permissions);
 		return (bool)$check;
 	}
@@ -1105,9 +1240,10 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public function canViewStage($stage = 'Live', $member = null) {
 		$oldMode = Versioned::get_reading_mode();
-		Versioned::reading_stage($stage);
+		Versioned::set_stage($stage);
 
-		$versionFromStage = DataObject::get($this->owner->class)->byID($this->owner->ID);
+		$owner = $this->owner;
+		$versionFromStage = DataObject::get($owner->class)->byID($owner->ID);
 
 		Versioned::set_reading_mode($oldMode);
 		return $versionFromStage ? $versionFromStage->canView($member) : false;
@@ -1134,15 +1270,14 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return boolean Returns false if the field isn't in the table, true otherwise
 	 */
 	public function hasVersionField($table) {
-		$rPos = strrpos($table,'_');
-
-		if(($rPos !== false) && in_array(substr($table,$rPos), $this->stages)) {
-			$tableWithoutStage = substr($table,0,$rPos);
-		} else {
-			$tableWithoutStage = $table;
+		// Strip "_Live" from end of table
+		$live = static::LIVE;
+		if($this->hasStages() && preg_match("/^(?<table>.*)_{$live}$/", $table, $matches)) {
+			$table = $matches['table'];
 		}
 
-		return ('DataObject' == get_parent_class($tableWithoutStage));
+		// Base table has version field
+		return $table === ClassInfo::baseDataClass($table);
 	}
 
 	/**
@@ -1175,15 +1310,14 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public function latestPublished() {
 		// Get the root data object class - this will have the version field
-		$table1 = $this->owner->class;
-		while( ($p = get_parent_class($table1)) != "DataObject") $table1 = $p;
-
-		$table2 = $table1 . "_$this->liveStage";
+		$owner = $this->owner;
+		$table1 = ClassInfo::baseDataClass($owner);
+		$table2 = $this->stageTable($table1, static::LIVE);
 
 		return DB::prepared_query("SELECT \"$table1\".\"Version\" = \"$table2\".\"Version\" FROM \"$table1\"
 			 INNER JOIN \"$table2\" ON \"$table1\".\"ID\" = \"$table2\".\"ID\"
 			 WHERE \"$table1\".\"ID\" = ?",
-			array($this->owner->ID)
+			array($owner->ID)
 		)->value();
 	}
 
@@ -1194,14 +1328,102 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public function doPublish() {
 		$owner = $this->owner;
+		if(!$owner->canPublish()) {
+			return false;
+		}
+
 		$owner->invokeWithExtensions('onBeforePublish');
 		$owner->write();
-		$owner->publish("Stage", "Live");
+		$owner->publish(static::DRAFT, static::LIVE);
 		$owner->invokeWithExtensions('onAfterPublish');
 		return true;
 	}
 
+	/**
+	 * Trigger publishing of owned objects
+	 */
+	public function onAfterPublish() {
+		$owner = $this->owner;
 
+		// Publish owned objects
+		foreach ($owner->findOwned(false) as $object) {
+			/** @var Versioned|DataObject $object */
+			$object->doPublish();
+		}
+
+		// Unlink any objects disowned as a result of this action
+		// I.e. objects which aren't owned anymore by this record, but are by the old live record
+		$owner->unlinkDisownedObjects(Versioned::DRAFT, Versioned::LIVE);
+	}
+
+	/**
+	 * Set foreign keys of has_many objects to 0 where those objects were
+	 * disowned as a result of a partial publish / unpublish.
+	 * I.e. this object and its owned objects were recently written to $targetStage,
+	 * but deleted objects were not.
+	 *
+	 * Note that this operation does not create any new Versions
+	 *
+	 * @param string $sourceStage Objects in this stage will not be unlinked.
+	 * @param string $targetStage Objects which exist in this stage but not $sourceStage
+	 * will be unlinked.
+	 */
+	public function unlinkDisownedObjects($sourceStage, $targetStage) {
+		$owner = $this->owner;
+
+		// after publishing, objects which used to be owned need to be
+		// dis-connected from this object (set ForeignKeyID = 0)
+		$owns = $owner->config()->owns;
+		$hasMany = $owner->config()->has_many;
+		if(empty($owns) || empty($hasMany)) {
+			return;
+		}
+
+		$ownedHasMany = array_intersect($owns, array_keys($hasMany));
+		foreach($ownedHasMany as $relationship) {
+			// Find metadata on relationship
+			$joinClass = $owner->hasManyComponent($relationship);
+			$joinField = $owner->getRemoteJoinField($relationship, 'has_many', $polymorphic);
+			$idField = $polymorphic ? "{$joinField}ID" : $joinField;
+			$joinTable = ClassInfo::table_for_object_field($joinClass, $idField);
+
+			// Generate update query which will unlink disowned objects
+			$targetTable = $this->stageTable($joinTable, $targetStage);
+			$disowned = new SQLUpdate("\"{$targetTable}\"");
+			$disowned->assign("\"{$idField}\"", 0);
+			$disowned->addWhere(array(
+				"\"{$targetTable}\".\"{$idField}\"" => $owner->ID
+			));
+
+			// Build exclusion list (items to owned objects we need to keep)
+			$sourceTable = $this->stageTable($joinTable, $sourceStage);
+			$owned = new SQLSelect("\"{$sourceTable}\".\"ID\"", "\"{$sourceTable}\"");
+			$owned->addWhere(array(
+				"\"{$sourceTable}\".\"{$idField}\"" => $owner->ID
+			));
+
+			// Apply class condition if querying on polymorphic has_one
+			if($polymorphic) {
+				$disowned->assign("\"{$joinField}Class\"", null);
+				$disowned->addWhere(array(
+					"\"{$targetTable}\".\"{$joinField}Class\"" => get_class($owner)
+				));
+				$owned->addWhere(array(
+					"\"{$sourceTable}\".\"{$joinField}Class\"" => get_class($owner)
+				));
+			}
+
+			// Merge queries and perform unlink
+			$ownedSQL = $owned->sql($ownedParams);
+			$disowned->addWhere(array(
+				"\"{$targetTable}\".\"ID\" NOT IN ({$ownedSQL})" => $ownedParams
+			));
+
+			$owner->extend('updateDisownershipQuery', $disowned, $sourceStage, $targetStage, $relationship);
+
+			$disowned->execute();
+		}
+	}
 
 	/**
 	 * Removes the record from both live and stage
@@ -1210,54 +1432,105 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public function doArchive() {
 		$owner = $this->owner;
-		$owner->invokeWithExtensions('onBeforeArchive', $this);
+		if(!$owner->canArchive()) {
+			return false;
+		}
 
-		if($owner->doUnpublish()) {
+		$owner->invokeWithExtensions('onBeforeArchive', $this);
+		$owner->doUnpublish();
 		$owner->delete();
 		$owner->invokeWithExtensions('onAfterArchive', $this);
 
 		return true;
 	}
 
-		return false;
-	}
-
 	/**
 	 * Removes this record from the live site
 	 *
 	 * @return bool Flag whether the unpublish was successful
-	 *
-	 * @uses SiteTreeExtension->onBeforeUnpublish()
-	 * @uses SiteTreeExtension->onAfterUnpublish()
 	 */
 	public function doUnpublish() {
 		$owner = $this->owner;
+		if(!$owner->canUnpublish()) {
+			return false;
+		}
+
+		// Skip if this record isn't saved
 		if(!$owner->isInDB()) {
 			return false;
 		}
 
+		// Skip if this record isn't on live
+		if(!$owner->isPublished()) {
+			return false;
+		}
+
 		$owner->invokeWithExtensions('onBeforeUnpublish');
 
-		$origStage = self::current_stage();
-		self::reading_stage(self::get_live_stage());
+		$origStage = static::get_stage();
+		static::set_stage(static::LIVE);
 
 		// This way our ID won't be unset
 		$clone = clone $owner;
 		$clone->delete();
 
-		self::reading_stage($origStage);
-
-		// If we're on the draft site, then we can update the status.
-		// Otherwise, these lines will resurrect an inappropriate record
-		if(self::current_stage() != self::get_live_stage() && $this->isOnDraft()) {
-			$owner->write();
-		}
+		static::set_stage($origStage);
 
 		$owner->invokeWithExtensions('onAfterUnpublish');
-
 		return true;
 	}
 
+	/**
+	 * Trigger unpublish of owning objects
+	 */
+	public function onAfterUnpublish() {
+		$owner = $this->owner;
+
+		// Any objects which owned (and thus relied on the unpublished object) are now unpublished automatically.
+		foreach ($owner->findOwners(false) as $object) {
+			/** @var Versioned|DataObject $object */
+			$object->doUnpublish();
+		}
+	}
+
+
+	/**
+	 * Revert the draft changes: replace the draft content with the content on live
+	 *
+	 * @return bool True if the revert was successful
+	 */
+	public function doRevertToLive() {
+		$owner = $this->owner;
+		if(!$owner->canRevertToLive()) {
+			return false;
+		}
+
+		$owner->invokeWithExtensions('onBeforeRevertToLive');
+		$owner->publish("Live", "Stage", false);
+		$owner->invokeWithExtensions('onAfterRevertToLive');
+		return true;
+	}
+
+	/**
+	 * Trigger revert of all owned objects to stage
+	 */
+	public function onAfterRevertToLive() {
+		$owner = $this->owner;
+		/** @var Versioned|DataObject $liveOwner */
+		$liveOwner = static::get_by_stage(get_class($owner), static::LIVE)
+			->byID($owner->ID);
+
+		// Revert any owned objects from the live stage only
+		foreach ($liveOwner->findOwned(false) as $object) {
+			/** @var Versioned|DataObject $object */
+			$object->doRevertToLive();
+		}
+
+		// Unlink any objects disowned as a result of this action
+		// I.e. objects which aren't owned anymore by this record, but are by the old draft record
+		$owner->unlinkDisownedObjects(Versioned::LIVE, Versioned::DRAFT);
+	}
+
 	/**
 	 * Move a database record from one stage to the other.
 	 *
@@ -1276,7 +1549,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		if(is_numeric($fromStage)) {
 			$from = Versioned::get_version($baseClass, $owner->ID, $fromStage);
 		} else {
-			$this->owner->flushCache();
+			$owner->flushCache();
 			$from = Versioned::get_one_by_stage($baseClass, $fromStage, array(
 				"\"{$baseClass}\".\"ID\" = ?" => $owner->ID
 			));
@@ -1304,7 +1577,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 
 		// Change to new stage, write, and revert state
 		$oldMode = Versioned::get_reading_mode();
-		Versioned::reading_stage($toStage);
+		Versioned::set_stage($toStage);
 
 		// Migrate stage prior to write
 		$from->setSourceQueryParam('Versioned.mode', 'stage');
@@ -1342,12 +1615,14 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 *
 	 * @param string $stage1 The first stage to check.
 	 * @param string $stage2
+	 * @return bool
 	 */
 	public function stagesDiffer($stage1, $stage2) {
 		$table1 = $this->baseTable($stage1);
 		$table2 = $this->baseTable($stage2);
 
-		if(!is_numeric($this->owner->ID)) {
+		$owner = $this->owner;
+		if(!is_numeric($owner->ID)) {
 			return true;
 		}
 
@@ -1359,7 +1634,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 			"SELECT CASE WHEN \"$table1\".\"Version\"=\"$table2\".\"Version\" THEN 1 ELSE 0 END
 			 FROM \"$table1\" INNER JOIN \"$table2\" ON \"$table1\".\"ID\" = \"$table2\".\"ID\"
 			 AND \"$table1\".\"ID\" = ?",
-			array($this->owner->ID)
+			array($owner->ID)
 		)->value();
 
 		return !$stagesAreEqual;
@@ -1371,6 +1646,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @param string $limit
 	 * @param string $join Deprecated, use leftJoin($table, $joinClause) instead
 	 * @param string $having
+	 * @return ArrayList
 	 */
 	public function Versions($filter = "", $sort = "", $limit = "", $join = "", $having = "") {
 		return $this->allVersions($filter, $sort, $limit, $join, $having);
@@ -1388,11 +1664,14 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public function allVersions($filter = "", $sort = "", $limit = "", $join = "", $having = "") {
 		// Make sure the table names are not postfixed (e.g. _Live)
-		$oldMode = self::get_reading_mode();
-		self::reading_stage('Stage');
+		$oldMode = static::get_reading_mode();
+		static::set_stage('Stage');
 
-		$list = DataObject::get(get_class($this->owner), $filter, $sort, $join, $limit);
-		if($having) $having = $list->having($having);
+		$owner = $this->owner;
+		$list = DataObject::get(get_class($owner), $filter, $sort, $join, $limit);
+		if($having) {
+			$list->having($having);
+		}
 
 		$query = $list->dataQuery()->query();
 
@@ -1414,7 +1693,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		}
 
 		$query->addWhere(array(
-			"\"{$baseTable}_versions\".\"RecordID\" = ?" => $this->owner->ID
+			"\"{$baseTable}_versions\".\"RecordID\" = ?" => $owner->ID
 		));
 		$query->setOrderBy(($sort) ? $sort
 			: "\"{$baseTable}_versions\".\"LastEdited\" DESC, \"{$baseTable}_versions\".\"Version\" DESC");
@@ -1439,8 +1718,9 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return DataObject
 	 */
 	public function compareVersions($from, $to) {
-		$fromRecord = Versioned::get_version($this->owner->class, $this->owner->ID, $from);
-		$toRecord = Versioned::get_version($this->owner->class, $this->owner->ID, $to);
+		$owner = $this->owner;
+		$fromRecord = Versioned::get_version($owner->class, $owner->ID, $from);
+		$toRecord = Versioned::get_version($owner->class, $owner->ID, $to);
 
 		$diff = new DataDifferencer($fromRecord, $toRecord);
 
@@ -1454,14 +1734,24 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return string
 	 */
 	public function baseTable($stage = null) {
-		$tableClasses = ClassInfo::dataClassesFor($this->owner->class);
-		$baseClass = array_shift($tableClasses);
-
-		if(!$stage || $stage == $this->defaultStage) {
-			return $baseClass;
+		$baseClass = ClassInfo::baseDataClass($this->owner);
+		return $this->stageTable($baseClass, $stage);
 	}
 
-		return $baseClass . "_$stage";
+	/**
+	 * Given a class and stage determine the table name.
+	 *
+	 * Note: Stages this asset does not exist in will default to the draft table.
+	 *
+	 * @param string $class
+	 * @param string $stage
+	 * @return string Table name
+	 */
+	public function stageTable($class, $stage) {
+		if($this->hasStages() && $stage === static::LIVE) {
+			return "{$class}_{$stage}";
+		}
+		return $class;
 	}
 
 	//-----------------------------------------------------------------------------------------------//
@@ -1475,7 +1765,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public static function can_choose_site_stage($request) {
 		// Request is allowed if stage isn't being modified
-		if((!$request->getVar('stage') || $request->getVar('stage') === static::get_live_stage())
+		if((!$request->getVar('stage') || $request->getVar('stage') === static::LIVE)
 			&& !$request->getVar('archiveDate')
 		) {
 			return true;
@@ -1506,14 +1796,16 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 		// Determine the reading mode
 		if(isset($_GET['stage'])) {
 			$stage = ucfirst(strtolower($_GET['stage']));
-			if(!in_array($stage, array('Stage', 'Live'))) $stage = 'Live';
+			if(!in_array($stage, array(static::DRAFT, static::LIVE))) {
+				$stage = static::LIVE;
+			}
 			$mode = 'Stage.' . $stage;
 		} elseif (isset($_GET['archiveDate']) && strtotime($_GET['archiveDate'])) {
 			$mode = 'Archive.' . $_GET['archiveDate'];
 		} elseif($preexistingMode) {
 			$mode = $preexistingMode;
 		} else {
-			$mode = self::DEFAULT_MODE;
+			$mode = static::DEFAULT_MODE;
 		}
 
 		// Save reading mode
@@ -1521,13 +1813,13 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 
 		// Try not to store the mode in the session if not needed
 		if(($preexistingMode && $preexistingMode !== $mode)
-			|| (!$preexistingMode && $mode !== self::DEFAULT_MODE)
+			|| (!$preexistingMode && $mode !== static::DEFAULT_MODE)
 		) {
 			Session::set('readingMode', $mode);
 		}
 
 		if(!headers_sent() && !Director::is_cli()) {
-			if(Versioned::current_stage() == 'Live') {
+			if(Versioned::get_stage() == 'Live') {
 				// clear the cookie if it's set
 				if(Cookie::get('bypassStaticCache')) {
 					Cookie::force_expiry('bypassStaticCache', null, null, false, true /* httponly */);
@@ -1547,7 +1839,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @param string $mode
 	 */
 	public static function set_reading_mode($mode) {
-		Versioned::$reading_mode = $mode;
+		self::$reading_mode = $mode;
 	}
 
 	/**
@@ -1556,16 +1848,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return string
 	 */
 	public static function get_reading_mode() {
-		return Versioned::$reading_mode;
-	}
-
-	/**
-	 * Get the name of the 'live' stage.
-	 *
-	 * @return string
-	 */
-	public static function get_live_stage() {
-		return "Live";
+		return self::$reading_mode;
 	}
 
 	/**
@@ -1573,7 +1856,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 *
 	 * @return string
 	 */
-	public static function current_stage() {
+	public static function get_stage() {
 		$parts = explode('.', Versioned::get_reading_mode());
 
 		if($parts[0] == 'Stage') {
@@ -1588,7 +1871,9 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public static function current_archived_date() {
 		$parts = explode('.', Versioned::get_reading_mode());
-		if($parts[0] == 'Archive') return $parts[1];
+		if($parts[0] == 'Archive') {
+			return $parts[1];
+		}
 	}
 
 	/**
@@ -1596,8 +1881,8 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 *
 	 * @param string $stage New reading stage.
 	 */
-	public static function reading_stage($stage) {
-		Versioned::set_reading_mode('Stage.' . $stage);
+	public static function set_stage($stage) {
+		static::set_reading_mode('Stage.' . $stage);
 	}
 
 	/**
@@ -1623,7 +1908,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public static function get_one_by_stage($class, $stage, $filter = '', $cache = true, $sort = '') {
 		// TODO: No identity cache operating
-		$items = self::get_by_stage($class, $stage, $filter, $sort, null, 1);
+		$items = static::get_by_stage($class, $stage, $filter, $sort, null, 1);
 
 		return $items->First();
 	}
@@ -1736,14 +2021,15 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public function deleteFromStage($stage) {
 		$oldMode = Versioned::get_reading_mode();
-		Versioned::reading_stage($stage);
-		$clone = clone $this->owner;
+		Versioned::set_stage($stage);
+		$owner = $this->owner;
+		$clone = clone $owner;
 		$clone->delete();
 		Versioned::set_reading_mode($oldMode);
 
 		// Fix the version number cache (in case you go delete from stage and then check ExistsOnLive)
-		$baseClass = ClassInfo::baseDataClass($this->owner->class);
-		self::$cache_versionnumber[$baseClass][$stage][$this->owner->ID] = null;
+		$baseClass = ClassInfo::baseDataClass($owner->class);
+		self::$cache_versionnumber[$baseClass][$stage][$owner->ID] = null;
 	}
 
 	/**
@@ -1755,10 +2041,11 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 */
 	public function writeToStage($stage, $forceInsert = false) {
 		$oldMode = Versioned::get_reading_mode();
-		Versioned::reading_stage($stage);
+		Versioned::set_stage($stage);
 
-		$this->owner->forceChange();
-		$result = $this->owner->write(false, $forceInsert);
+		$owner = $this->owner;
+		$owner->forceChange();
+		$result = $owner->write(false, $forceInsert);
 		Versioned::set_reading_mode($oldMode);
 
 		return $result;
@@ -1768,16 +2055,31 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * Roll the draft version of this record to match the published record.
 	 * Caution: Doesn't overwrite the object properties with the rolled back version.
 	 *
-	 * @param int $version Either the string 'Live' or a version number
+	 * {@see doRevertToLive()} to reollback to live
+	 *
+	 * @param int $version Version number
 	 */
 	public function doRollbackTo($version) {
 		$owner = $this->owner;
 		$owner->extend('onBeforeRollback', $version);
-		$this->publish($version, "Stage", true);
+		$owner->publish($version, "Stage", true);
 		$owner->writeWithoutVersion();
 		$owner->extend('onAfterRollback', $version);
 	}
 
+	public function onAfterRollback($version) {
+		// Find record at this version
+		$baseClass = ClassInfo::baseDataClass($this->owner);
+		$recordVersion = static::get_version($baseClass, $this->owner->ID, $version);
+
+		// Note that unlike other publishing actions, rollback is NOT recursive;
+		// The owner collects all objects and writes them back using writeToStage();
+		foreach ($recordVersion->findOwned() as $object) {
+			/** @var Versioned|DataObject $object */
+			$object->writeToStage(static::DRAFT);
+		}
+	}
+
 	/**
 	 * Return the latest version of the given record.
 	 *
@@ -1788,10 +2090,9 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	public static function get_latest_version($class, $id) {
 		$baseClass = ClassInfo::baseDataClass($class);
 		$list = DataList::create($baseClass)
-			->where(array("\"$baseClass\".\"RecordID\"" => $id))
 			->setDataQueryParam("Versioned.mode", "latest_versions");
 
-		return $list->First();
+		return $list->byID($id);
 	}
 
 	/**
@@ -1805,12 +2106,13 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return boolean
 	 */
 	public function isLatestVersion() {
-		if(!$this->owner->isInDB()) {
+		$owner = $this->owner;
+		if(!$owner->isInDB()) {
 			return false;
 		}
 
-		$version = self::get_latest_version($this->owner->class, $this->owner->ID);
-		return ($version->Version == $this->owner->Version);
+		$version = static::get_latest_version($owner->class, $owner->ID);
+		return ($version->Version == $owner->Version);
 	}
 
 	/**
@@ -1819,14 +2121,21 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return bool
 	 */
 	public function isPublished() {
-		if(!$this->owner->isInDB()) {
+		$owner = $this->owner;
+		if(!$owner->isInDB()) {
 			return false;
 		}
 
-		$table = ClassInfo::baseDataClass($this->owner->class) . '_' . self::get_live_stage();
+		// Non-staged objects are considered "published" if saved
+		if(!$this->hasStages()) {
+			return true;
+		}
+
+		$baseClass = ClassInfo::baseDataClass($owner->class);
+		$table = $this->stageTable($baseClass, static::LIVE);
 		$result = DB::prepared_query(
 			"SELECT COUNT(*) FROM \"{$table}\" WHERE \"{$table}\".\"ID\" = ?",
-			array($this->owner->ID)
+			array($owner->ID)
 		);
 		return (bool)$result->value();
 	}
@@ -1837,14 +2146,15 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return bool
 	 */
 	public function isOnDraft() {
-		if(!$this->owner->isInDB()) {
+		$owner = $this->owner;
+		if(!$owner->isInDB()) {
 			return false;
 		}
 
-		$table = ClassInfo::baseDataClass($this->owner->class);
+		$table = ClassInfo::baseDataClass($owner->class);
 		$result = DB::prepared_query(
 			"SELECT COUNT(*) FROM \"{$table}\" WHERE \"{$table}\".\"ID\" = ?",
-			array($this->owner->ID)
+			array($owner->ID)
 		);
 		return (bool)$result->value();
 	}
@@ -1887,13 +2197,12 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	public static function get_version($class, $id, $version) {
 		$baseClass = ClassInfo::baseDataClass($class);
 		$list = DataList::create($baseClass)
-			->where(array(
-				"\"{$baseClass}\".\"RecordID\"" => $id,
-				"\"{$baseClass}\".\"Version\"" => $version
-			))
-			->setDataQueryParam("Versioned.mode", 'all_versions');
+			->setDataQueryParam([
+				"Versioned.mode" => 'version',
+				"Versioned.version" => $version
+			]);
 
-		return $list->First();
+		return $list->byID($id);
 	}
 
 	/**
@@ -1948,7 +2257,7 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return string
 	 */
 	public function cacheKeyComponent() {
-		return 'versionedmode-'.self::get_reading_mode();
+		return 'versionedmode-'.static::get_reading_mode();
 	}
 
 	/**
@@ -1957,14 +2266,11 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 	 * @return array
 	 */
 	public function getVersionedStages() {
-		return $this->stages;
+		if($this->hasStages()) {
+			return [static::DRAFT, static::LIVE];
+		} else {
+			return [static::DRAFT];
 		}
-
-	/**
-	 * @return string
-	 */
-	public function getDefaultStage() {
-		return $this->defaultStage;
 	}
 
 	public static function get_template_global_variables() {
@@ -1972,6 +2278,15 @@ class Versioned extends DataExtension implements TemplateGlobalProvider {
 			'CurrentReadingMode' => 'get_reading_mode'
 		);
 	}
+
+	/**
+	 * Check if this object has stages
+	 *
+	 * @return bool True if this object is staged
+	 */
+	public function hasStages() {
+		return $this->mode === static::STAGEDVERSIONED;
+	}
 }
 
 /**

tests/filesystem/AssetControlExtensionTest.php

@@ -15,7 +15,7 @@ class AssetControlExtensionTest extends SapphireTest {
 		parent::setUp();
 
 		// Set backend and base url
-		\Versioned::reading_stage('Stage');
+		\Versioned::set_stage(Versioned::DRAFT);
 		AssetStoreTest_SpyStore::activate('AssetControlExtensionTest');
 		$this->logInWithPermission('ADMIN');
 
@@ -46,7 +46,7 @@ class AssetControlExtensionTest extends SapphireTest {
 	}
 
 	public function testFileDelete() {
-		\Versioned::reading_stage('Stage');
+		\Versioned::set_stage(Versioned::DRAFT);
 
 		/** @var AssetControlExtensionTest_VersionedObject $object1 */
 		$object1 = AssetControlExtensionTest_VersionedObject::get()
@@ -120,7 +120,7 @@ class AssetControlExtensionTest extends SapphireTest {
 	 * Test files being replaced
 	 */
 	public function testReplaceFile() {
-		\Versioned::reading_stage('Stage');
+		\Versioned::set_stage(Versioned::DRAFT);
 
 		/** @var AssetControlExtensionTest_VersionedObject $object1 */
 		$object1 = AssetControlExtensionTest_VersionedObject::get()

tests/filesystem/AssetStoreTest.php

@@ -602,6 +602,7 @@ class AssetStoreTest_SpyStore extends FlysystemAssetStore {
 	 * Helper method to get local filesystem path for this file
 	 *
 	 * @param AssetContainer $asset
+	 * @return string
 	 */
 	public static function getLocalPath(AssetContainer $asset) {
 		if($asset instanceof Folder) {
@@ -621,7 +622,6 @@ class AssetStoreTest_SpyStore extends FlysystemAssetStore {
 		/** @var Local $adapter */
 		$adapter = $filesystem->getAdapter();
 		return $adapter->applyPathPrefix($fileID);
-
 	}
 
 	public function cleanFilename($filename) {

tests/filesystem/FileMigrationHelperTest.php

@@ -21,6 +21,7 @@ class FileMigrationHelperTest extends SapphireTest {
 	 * @return string
 	 */
 	protected function getBasePath() {
+		// Note that the actual filesystem base is the 'assets' subdirectory within this
 		return ASSETS_PATH . '/FileMigrationHelperTest';
 	}
 
@@ -36,7 +37,7 @@ class FileMigrationHelperTest extends SapphireTest {
 		// Ensure that each file has a local record file in this new assets base
 		$from = FRAMEWORK_PATH . '/tests/model/testimages/test-image-low-quality.jpg';
 		foreach(File::get()->exclude('ClassName', 'Folder') as $file) {
-			$dest = $this->getBasePath() . '/assets/' . $file->getFilename();
+			$dest = AssetStoreTest_SpyStore::base_path() . '/' . $file->generateFilename();
 			SS_Filesystem::makeFolder(dirname($dest));
 			copy($from, $dest);
 		}
@@ -55,7 +56,7 @@ class FileMigrationHelperTest extends SapphireTest {
 	public function testMigration() {
 		// Prior to migration, check that each file has empty Filename / Hash properties
 		foreach(File::get()->exclude('ClassName', 'Folder') as $file) {
-			$filename = $file->getFilename();
+			$filename = $file->generateFilename();
 			$this->assertNotEmpty($filename, "File {$file->Name} has a filename");
 			$this->assertEmpty($file->File->getFilename(), "File {$file->Name} has no DBFile filename");
 			$this->assertEmpty($file->File->getHash(), "File {$file->Name} has no hash");
@@ -70,20 +71,25 @@ class FileMigrationHelperTest extends SapphireTest {
 
 		// Test that each file exists
 		foreach(File::get()->exclude('ClassName', 'Folder') as $file) {
+			$expectedFilename = $file->generateFilename();
 			$filename = $file->File->getFilename();
+			$this->assertTrue($file->exists(), "File with name {$filename} exists");
 			$this->assertNotEmpty($filename, "File {$file->Name} has a Filename");
+			$this->assertEquals($expectedFilename, $filename, "File {$file->Name} has retained its Filename value");
 			$this->assertEquals(
 				'33be1b95cba0358fe54e8b13532162d52f97421c',
 				$file->File->getHash(),
 				"File with name {$filename} has the correct hash"
 			);
-			$this->assertTrue($file->exists(), "File with name {$filename} exists");
 			$this->assertTrue($file->isPublished(), "File is published after migration");
 		}
 	}
 
 }
 
+/**
+ * @property File $owner
+ */
 class FileMigrationHelperTest_Extension extends DataExtension implements TestOnly {
 	/**
 	 * Ensure that File dataobject has the legacy "Filename" field
@@ -94,6 +100,6 @@ class FileMigrationHelperTest_Extension extends DataExtension implements TestOnl
 
 	public function onBeforeWrite() {
 		// Ensure underlying filename field is written to the database
-		$this->owner->setField('Filename', 'assets/' . $this->owner->getFilename());
+		$this->owner->setField('Filename', 'assets/' . $this->owner->generateFilename());
 	}
 }

tests/filesystem/FileTest.php

@@ -15,7 +15,7 @@ class FileTest extends SapphireTest {
 	public function setUp() {
 		parent::setUp();
 		$this->logInWithPermission('ADMIN');
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 
 		// Set backend root to /ImageTest
 		AssetStoreTest_SpyStore::activate('FileTest');
@@ -242,7 +242,7 @@ class FileTest extends SapphireTest {
 		// Rename
 		$file->Name = 'renamed.txt';
 		$newTuple = $oldTuple;
-		$newTuple['Filename'] = $file->getFilename();
+		$newTuple['Filename'] = $file->generateFilename();
 
 		// Before write()
 		$this->assertTrue(
@@ -287,7 +287,7 @@ class FileTest extends SapphireTest {
 		// set ParentID
 		$file->ParentID = $subfolder->ID;
 		$newTuple = $oldTuple;
-		$newTuple['Filename'] = $file->getFilename();
+		$newTuple['Filename'] = $file->generateFilename();
 
 		// Before write()
 		$this->assertTrue(

tests/filesystem/FolderTest.php

@@ -16,7 +16,7 @@ class FolderTest extends SapphireTest {
 		parent::setUp();
 
 		$this->logInWithPermission('ADMIN');
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 
 		// Set backend root to /FolderTest
 		AssetStoreTest_SpyStore::activate('FolderTest');
@@ -110,25 +110,44 @@ class FolderTest extends SapphireTest {
 		Folder::find_or_make($folder1->Filename);
 		$folder2 = $this->objFromFixture('Folder', 'folder2');
 
+		// Publish file1
+		$file1 = DataObject::get_by_id('File', $this->idFromFixture('File', 'file1-folder1'), false);
+		$file1->doPublish();
+
 		// set ParentID. This should cause updateFilesystem to be called on all children
 		$folder1->ParentID = $folder2->ID;
 		$folder1->write();
 
 		// Check if the file in the folder moved along
-		$file1 = DataObject::get_by_id('File', $this->idFromFixture('File', 'file1-folder1'), false);
-		$this->assertFileExists(AssetStoreTest_SpyStore::getLocalPath($file1));
+		$file1Draft = Versioned::get_by_stage('File', Versioned::DRAFT)->byID($file1->ID);
+		$this->assertFileExists(AssetStoreTest_SpyStore::getLocalPath($file1Draft));
 
 		$this->assertEquals(
 			'FileTest-folder2/FileTest-folder1/File1.txt',
-			$file1->Filename,
+			$file1Draft->Filename,
 			'The file DataObject has updated path'
 		);
 
 		// File should be located in new folder
 		$this->assertEquals(
 			ASSETS_PATH . '/FolderTest/.protected/FileTest-folder2/FileTest-folder1/55b443b601/File1.txt',
-			AssetStoreTest_SpyStore::getLocalPath($file1)
+			AssetStoreTest_SpyStore::getLocalPath($file1Draft)
 		);
+
+		// Published (live) version remains in the old location
+		$file1Live = Versioned::get_by_stage('File', Versioned::LIVE)->byID($file1->ID);
+		$this->assertEquals(
+			ASSETS_PATH . '/FolderTest/FileTest-folder1/55b443b601/File1.txt',
+			AssetStoreTest_SpyStore::getLocalPath($file1Live)
+		);
+
+		// Publishing the draft to live should move the new file to the public store
+		$file1Draft->doPublish();
+		$this->assertEquals(
+			ASSETS_PATH . '/FolderTest/FileTest-folder2/FileTest-folder1/55b443b601/File1.txt',
+			AssetStoreTest_SpyStore::getLocalPath($file1Draft)
+		);
+
 	}
 
 	/**

tests/filesystem/UploadTest.php

@@ -10,7 +10,7 @@ class UploadTest extends SapphireTest {
 
 	public function setUp() {
 		parent::setUp();
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 		AssetStoreTest_SpyStore::activate('UploadTest');
 	}
 

tests/forms/uploadfield/AssetFieldTest.php

@@ -16,7 +16,7 @@ class AssetFieldTest extends FunctionalTest {
 		parent::setUp();
 
 		$this->logInWithPermission('ADMIN');
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 
 		// Set backend root to /AssetFieldTest
 		AssetStoreTest_SpyStore::activate('AssetFieldTest');

tests/forms/uploadfield/UploadFieldTest.php

@@ -23,7 +23,7 @@ class UploadFieldTest extends FunctionalTest {
 
 		// Save versioned state
 		$this->oldReadingMode = Versioned::get_reading_mode();
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 
 		// Set backend root to /UploadFieldTest
 		AssetStoreTest_SpyStore::activate('UploadFieldTest');

tests/model/DataDifferencerTest.php

@@ -19,7 +19,7 @@ class DataDifferencerTest extends SapphireTest {
 	public function setUp() {
 		parent::setUp();
 
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 
 		// Set backend root to /DataDifferencerTest
 		AssetStoreTest_SpyStore::activate('DataDifferencerTest');

tests/model/DataObjectLazyLoadingTest.php

@@ -408,7 +408,7 @@ class DataObjectLazyLoadingTest extends SapphireTest {
 			$obj1ID
 		));
 
-		Versioned::reading_stage('Live');
+		Versioned::set_stage(Versioned::LIVE);
 		$obj1 = VersionedLazy_DataObject::get()->byID($obj1ID);
 		$this->assertEquals(
 			'live-value',

tests/model/VersionedOwnershipTest.php

@@ -10,6 +10,7 @@ class VersionedOwnershipTest extends SapphireTest {
 		'VersionedOwnershipTest_Subclass',
 		'VersionedOwnershipTest_Related',
 		'VersionedOwnershipTest_Attachment',
+		'VersionedOwnershipTest_RelatedMany',
 	);
 
 	protected static $fixture_file = 'VersionedOwnershipTest.yml';
@@ -18,7 +19,7 @@ class VersionedOwnershipTest extends SapphireTest {
 	{
 		parent::setUp();
 
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 
 		// Automatically publish any object named *_published
 		foreach($this->getFixtureFactory()->getFixtures() as $class => $fixtures) {
@@ -32,6 +33,18 @@ class VersionedOwnershipTest extends SapphireTest {
 		}
 	}
 
+	/**
+	 * Virtual "sleep" that doesn't actually slow execution, only advances SS_DateTime::now()
+	 *
+	 * @param int $minutes
+	 */
+	protected function sleep($minutes) {
+		$now = SS_Datetime::now();
+		$date = DateTime::createFromFormat('Y-m-d H:i:s', $now->getValue());
+		$date->modify("+{$minutes} minutes");
+		SS_Datetime::set_mock_now($date->format('Y-m-d H:i:s'));
+	}
+
 	/**
 	 * Test basic findOwned() in stage mode
 	 */
@@ -44,6 +57,9 @@ class VersionedOwnershipTest extends SapphireTest {
 				['Title' => 'Attachment 1'],
 				['Title' => 'Attachment 2'],
 				['Title' => 'Attachment 5'],
+				['Title' => 'Related Many 1'],
+				['Title' => 'Related Many 2'],
+				['Title' => 'Related Many 3'],
 			],
 			$subclass1->findOwned()
 		);
@@ -52,6 +68,9 @@ class VersionedOwnershipTest extends SapphireTest {
 		$this->assertDOSEquals(
 			[
 				['Title' => 'Related 1'],
+				['Title' => 'Related Many 1'],
+				['Title' => 'Related Many 2'],
+				['Title' => 'Related Many 3'],
 			],
 			$subclass1->findOwned(false)
 		);
@@ -64,6 +83,7 @@ class VersionedOwnershipTest extends SapphireTest {
 				['Title' => 'Attachment 3'],
 				['Title' => 'Attachment 4'],
 				['Title' => 'Attachment 5'],
+				['Title' => 'Related Many 4'],
 			],
 			$subclass2->findOwned()
 		);
@@ -71,7 +91,8 @@ class VersionedOwnershipTest extends SapphireTest {
 		// Non-recursive search
 		$this->assertDOSEquals(
 			[
-				['Title' => 'Related 2']
+				['Title' => 'Related 2'],
+				['Title' => 'Related Many 4'],
 			],
 			$subclass2->findOwned(false)
 		);
@@ -175,6 +196,7 @@ class VersionedOwnershipTest extends SapphireTest {
 				['Title' => 'Related 2 Modified'],
 				['Title' => 'Attachment 3 Modified'],
 				['Title' => 'Attachment 5'],
+				['Title' => 'Related Many 4'],
 			],
 			$subclass2Stage->findOwned()
 		);
@@ -183,6 +205,7 @@ class VersionedOwnershipTest extends SapphireTest {
 		$this->assertDOSEquals(
 			[
 				['Title' => 'Related 2 Modified'],
+				['Title' => 'Related Many 4'],
 			],
 			$subclass2Stage->findOwned(false)
 		);
@@ -196,6 +219,7 @@ class VersionedOwnershipTest extends SapphireTest {
 				['Title' => 'Attachment 3'],
 				['Title' => 'Attachment 4'],
 				['Title' => 'Attachment 5'],
+				['Title' => 'Related Many 4'],
 			],
 			$subclass2Live->findOwned()
 		);
@@ -204,10 +228,282 @@ class VersionedOwnershipTest extends SapphireTest {
 		$this->assertDOSEquals(
 			[
 				['Title' => 'Related 2'],
+				['Title' => 'Related Many 4'],
 			],
 			$subclass2Live->findOwned(false)
 		);
 	}
+
+	/**
+	 * Test that objects are correctly published recursively
+	 */
+	public function testRecursivePublish() {
+		/** @var VersionedOwnershipTest_Subclass $parent */
+		$parent = $this->objFromFixture('VersionedOwnershipTest_Subclass', 'subclass1_published');
+		$parentID = $parent->ID;
+		$banner1 = $this->objFromFixture('VersionedOwnershipTest_RelatedMany', 'relatedmany1_published');
+		$banner2 = $this->objFromFixture('VersionedOwnershipTest_RelatedMany', 'relatedmany2_published');
+		$banner2ID = $banner2->ID;
+
+		// Modify, Add, and Delete banners on stage
+		$banner1->Title = 'Renamed Banner 1';
+		$banner1->write();
+
+		$banner2->delete();
+
+		$banner4 = new VersionedOwnershipTest_RelatedMany();
+		$banner4->Title = 'New Banner';
+		$parent->Banners()->add($banner4);
+
+		// Check state of objects before publish
+		$oldLiveBanners = [
+			['Title' => 'Related Many 1'],
+			['Title' => 'Related Many 2'], // Will be deleted
+			// `Related Many 3` isn't published
+		];
+		$newBanners = [
+			['Title' => 'Renamed Banner 1'], // Renamed
+			['Title' => 'Related Many 3'], // Published without changes
+			['Title' => 'New Banner'], // Created
+		];
+		$parentDraft = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::DRAFT)
+			->byID($parentID);
+		$this->assertDOSEquals($newBanners, $parentDraft->Banners());
+		$parentLive = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::LIVE)
+			->byID($parentID);
+		$this->assertDOSEquals($oldLiveBanners, $parentLive->Banners());
+
+		// On publishing of owner, all children should now be updated
+		$parent->doPublish();
+
+		// Now check each object has the correct state
+		$parentDraft = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::DRAFT)
+			->byID($parentID);
+		$this->assertDOSEquals($newBanners, $parentDraft->Banners());
+		$parentLive = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::LIVE)
+			->byID($parentID);
+		$this->assertDOSEquals($newBanners, $parentLive->Banners());
+
+		// Check that the deleted banner hasn't actually been deleted from the live stage,
+		// but in fact has been unlinked.
+		$banner2Live = Versioned::get_by_stage('VersionedOwnershipTest_RelatedMany', Versioned::LIVE)
+			->byID($banner2ID);
+		$this->assertEmpty($banner2Live->PageID);
+	}
+
+	/**
+	 * Test that owning objects get unpublished as needed
+	 */
+	public function testRecursiveUnpublish() {
+		// Unsaved objects can't be unpublished
+		$unsaved = new VersionedOwnershipTest_Subclass();
+		$this->assertFalse($unsaved->doUnpublish());
+
+		// Draft-only objects can't be unpublished
+		/** @var VersionedOwnershipTest_RelatedMany $banner3Unpublished */
+		$banner3Unpublished = $this->objFromFixture('VersionedOwnershipTest_RelatedMany', 'relatedmany3');
+		$this->assertFalse($banner3Unpublished->doUnpublish());
+
+		// First test: mid-level unpublish; We expect that owners should be unpublished, but not
+		// owned objects, nor other siblings shared by the same owner.
+		$related2 = $this->objFromFixture('VersionedOwnershipTest_Related', 'related2_published');
+		/** @var VersionedOwnershipTest_Attachment $attachment3 */
+		$attachment3 = $this->objFromFixture('VersionedOwnershipTest_Attachment', 'attachment3_published');
+		/** @var VersionedOwnershipTest_RelatedMany $relatedMany4 */
+		$relatedMany4 = $this->objFromFixture('VersionedOwnershipTest_RelatedMany', 'relatedmany4_published');
+		/** @var VersionedOwnershipTest_Related $related2 */
+		$this->assertTrue($related2->doUnpublish());
+		$subclass2 = $this->objFromFixture('VersionedOwnershipTest_Subclass', 'subclass2_published');
+
+		/** @var VersionedOwnershipTest_Subclass $subclass2 */
+		$this->assertFalse($subclass2->isPublished()); // Owner IS unpublished
+		$this->assertTrue($attachment3->isPublished()); // Owned object is NOT unpublished
+		$this->assertTrue($relatedMany4->isPublished()); // Owned object by owner is NOT unpublished
+
+		// Second test: multi-level unpublish should recursively cascade down all owning objects
+		// Publish related2 again
+		$subclass2->doPublish();
+		$this->assertTrue($subclass2->isPublished());
+		$this->assertTrue($related2->isPublished());
+		$this->assertTrue($attachment3->isPublished());
+
+		// Unpublish leaf node
+		$this->assertTrue($attachment3->doUnpublish());
+
+		// Now all owning objects (only) are unpublished
+		$this->assertFalse($attachment3->isPublished()); // Unpublished because we just unpublished it
+		$this->assertFalse($related2->isPublished()); // Unpublished because it owns attachment3
+		$this->assertFalse($subclass2->isPublished()); // Unpublished ecause it owns related2
+		$this->assertTrue($relatedMany4->isPublished()); // Stays live because recursion only affects owners not owned.
+	}
+
+	public function testRecursiveArchive() {
+		// When archiving an object, any published owners should be unpublished at the same time
+		// but NOT achived
+
+		/** @var VersionedOwnershipTest_Attachment $attachment3 */
+		$attachment3 = $this->objFromFixture('VersionedOwnershipTest_Attachment', 'attachment3_published');
+		$attachment3ID = $attachment3->ID;
+		$this->assertTrue($attachment3->doArchive());
+
+		// This object is on neither stage nor live
+		$stageAttachment = Versioned::get_by_stage('VersionedOwnershipTest_Attachment', Versioned::DRAFT)
+			->byID($attachment3ID);
+		$liveAttachment = Versioned::get_by_stage('VersionedOwnershipTest_Attachment', Versioned::LIVE)
+			->byID($attachment3ID);
+		$this->assertEmpty($stageAttachment);
+		$this->assertEmpty($liveAttachment);
+
+		// Owning object is unpublished only
+		/** @var VersionedOwnershipTest_Related $stageOwner */
+		$stageOwner = $this->objFromFixture('VersionedOwnershipTest_Related', 'related2_published');
+		$this->assertTrue($stageOwner->isOnDraft());
+		$this->assertFalse($stageOwner->isPublished());
+
+		// Bottom level owning object is also unpublished
+		/** @var VersionedOwnershipTest_Subclass $stageTopOwner */
+		$stageTopOwner = $this->objFromFixture('VersionedOwnershipTest_Subclass', 'subclass2_published');
+		$this->assertTrue($stageTopOwner->isOnDraft());
+		$this->assertFalse($stageTopOwner->isPublished());
+	}
+
+	public function testRecursiveRevertToLive() {
+		/** @var VersionedOwnershipTest_Subclass $parent */
+		$parent = $this->objFromFixture('VersionedOwnershipTest_Subclass', 'subclass1_published');
+		$parentID = $parent->ID;
+		$banner1 = $this->objFromFixture('VersionedOwnershipTest_RelatedMany', 'relatedmany1_published');
+		$banner2 = $this->objFromFixture('VersionedOwnershipTest_RelatedMany', 'relatedmany2_published');
+		$banner2ID = $banner2->ID;
+
+		// Modify, Add, and Delete banners on stage
+		$banner1->Title = 'Renamed Banner 1';
+		$banner1->write();
+
+		$banner2->delete();
+
+		$banner4 = new VersionedOwnershipTest_RelatedMany();
+		$banner4->Title = 'New Banner';
+		$banner4->write();
+		$parent->Banners()->add($banner4);
+
+		// Check state of objects before publish
+		$liveBanners = [
+			['Title' => 'Related Many 1'],
+			['Title' => 'Related Many 2'],
+		];
+		$modifiedBanners = [
+			['Title' => 'Renamed Banner 1'], // Renamed
+			['Title' => 'Related Many 3'], // Published without changes
+			['Title' => 'New Banner'], // Created
+		];
+		$parentDraft = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::DRAFT)
+			->byID($parentID);
+		$this->assertDOSEquals($modifiedBanners, $parentDraft->Banners());
+		$parentLive = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::LIVE)
+			->byID($parentID);
+		$this->assertDOSEquals($liveBanners, $parentLive->Banners());
+
+		// When reverting parent, all records should be put back on stage
+		$this->assertTrue($parent->doRevertToLive());
+
+		// Now check each object has the correct state
+		$parentDraft = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::DRAFT)
+			->byID($parentID);
+		$this->assertDOSEquals($liveBanners, $parentDraft->Banners());
+		$parentLive = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::LIVE)
+			->byID($parentID);
+		$this->assertDOSEquals($liveBanners, $parentLive->Banners());
+
+		// Check that the newly created banner, even though it still exist, has been
+		// unlinked from the reverted draft record
+		/** @var VersionedOwnershipTest_RelatedMany $banner4Draft */
+		$banner4Draft = Versioned::get_by_stage('VersionedOwnershipTest_RelatedMany', Versioned::DRAFT)
+			->byID($banner4->ID);
+		$this->assertTrue($banner4Draft->isOnDraft());
+		$this->assertFalse($banner4Draft->isPublished());
+		$this->assertEmpty($banner4Draft->PageID);
+	}
+
+	/**
+	 * Test that rolling back to a single version works recursively
+	 */
+	public function testRecursiveRollback() {
+		/** @var VersionedOwnershipTest_Subclass $subclass2 */
+		$this->sleep(1);
+		$subclass2 = $this->objFromFixture('VersionedOwnershipTest_Subclass', 'subclass2_published');
+
+		// Create a few new versions
+		$versions = [];
+		for($version = 1; $version <= 3; $version++) {
+			// Write owned objects
+			$this->sleep(1);
+			foreach($subclass2->findOwned(true) as $obj) {
+				$obj->Title .= " - v{$version}";
+				$obj->write();
+			}
+			// Write parent
+			$this->sleep(1);
+			$subclass2->Title .= " - v{$version}";
+			$subclass2->write();
+			$versions[$version] = $subclass2->Version;
+		}
+
+
+		// Check reverting to first version
+		$this->sleep(1);
+		$subclass2->doRollbackTo($versions[1]);
+		/** @var VersionedOwnershipTest_Subclass $subclass2Draft */
+		$subclass2Draft = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::DRAFT)
+			->byID($subclass2->ID);
+		$this->assertEquals('Subclass 2 - v1', $subclass2Draft->Title);
+		$this->assertDOSEquals(
+			[
+				['Title' => 'Related 2 - v1'],
+				['Title' => 'Attachment 3 - v1'],
+				['Title' => 'Attachment 4 - v1'],
+				['Title' => 'Attachment 5 - v1'],
+				['Title' => 'Related Many 4 - v1'],
+			],
+			$subclass2Draft->findOwned(true)
+		);
+
+		// Check rolling forward to a later version
+		$this->sleep(1);
+		$subclass2->doRollbackTo($versions[3]);
+		/** @var VersionedOwnershipTest_Subclass $subclass2Draft */
+		$subclass2Draft = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::DRAFT)
+			->byID($subclass2->ID);
+		$this->assertEquals('Subclass 2 - v1 - v2 - v3', $subclass2Draft->Title);
+		$this->assertDOSEquals(
+			[
+				['Title' => 'Related 2 - v1 - v2 - v3'],
+				['Title' => 'Attachment 3 - v1 - v2 - v3'],
+				['Title' => 'Attachment 4 - v1 - v2 - v3'],
+				['Title' => 'Attachment 5 - v1 - v2 - v3'],
+				['Title' => 'Related Many 4 - v1 - v2 - v3'],
+			],
+			$subclass2Draft->findOwned(true)
+		);
+
+		// And rolling back one version
+		$this->sleep(1);
+		$subclass2->doRollbackTo($versions[2]);
+		/** @var VersionedOwnershipTest_Subclass $subclass2Draft */
+		$subclass2Draft = Versioned::get_by_stage('VersionedOwnershipTest_Subclass', Versioned::DRAFT)
+			->byID($subclass2->ID);
+		$this->assertEquals('Subclass 2 - v1 - v2', $subclass2Draft->Title);
+		$this->assertDOSEquals(
+			[
+				['Title' => 'Related 2 - v1 - v2'],
+				['Title' => 'Attachment 3 - v1 - v2'],
+				['Title' => 'Attachment 4 - v1 - v2'],
+				['Title' => 'Attachment 5 - v1 - v2'],
+				['Title' => 'Related Many 4 - v1 - v2'],
+			],
+			$subclass2Draft->findOwned(true)
+		);
+	}
+
 }
 
 /**
@@ -224,6 +520,11 @@ class VersionedOwnershipTest_Object extends DataObject implements TestOnly {
 	);
 }
 
+/**
+ * Object which:
+ * - owns a has_one object
+ * - owns has_many objects
+ */
 class VersionedOwnershipTest_Subclass extends VersionedOwnershipTest_Object implements TestOnly {
 	private static $db = array(
 		'Description' => 'Text',
@@ -233,12 +534,21 @@ class VersionedOwnershipTest_Subclass extends VersionedOwnershipTest_Object impl
 		'Related' => 'VersionedOwnershipTest_Related',
 	);
 
+	private static $has_many = array(
+		'Banners' => 'VersionedOwnershipTest_RelatedMany'
+	);
+
 	private static $owns = array(
 		'Related',
+		'Banners',
 	);
 }
 
 /**
+ * Object which:
+ * - owned by has_many objects
+ * - owns many_many Objects
+ *
  * @mixin Versioned
  */
 class VersionedOwnershipTest_Related extends DataObject implements TestOnly {
@@ -268,6 +578,29 @@ class VersionedOwnershipTest_Related extends DataObject implements TestOnly {
 	);
 }
 
+/**
+ * Object which is owned by a has_one object
+ *
+ * @mixin Versioned
+ */
+class VersionedOwnershipTest_RelatedMany extends DataObject implements TestOnly {
+	private static $extensions = array(
+		'Versioned',
+	);
+
+	private static $db = array(
+		'Title' => 'Varchar(255)',
+	);
+
+	private static $has_one = array(
+		'Page' => 'VersionedOwnershipTest_Subclass'
+	);
+
+	private static $owned_by = array(
+		'Page'
+	);
+}
+
 /**
  * @mixin Versioned
  */

tests/model/VersionedOwnershipTest.yml

@@ -26,6 +26,20 @@ VersionedOwnershipTest_Subclass:
     Title: 'Subclass 2'
     Related: =>VersionedOwnershipTest_Related.related2_published
 
+VersionedOwnershipTest_RelatedMany:
+  relatedmany1_published:
+    Title: 'Related Many 1'
+    Page: =>VersionedOwnershipTest_Subclass.subclass1_published
+  relatedmany2_published:
+    Title: 'Related Many 2'
+    Page: =>VersionedOwnershipTest_Subclass.subclass1_published
+  relatedmany3:
+    Title: 'Related Many 3'
+    Page: =>VersionedOwnershipTest_Subclass.subclass1_published
+  relatedmany4_published:
+    Title: 'Related Many 4'
+    Page: =>VersionedOwnershipTest_Subclass.subclass2_published
+
 VersionedOwnershipTest_Object:
   object1:
     Title: 'Object 1'

tests/model/VersionedTest.php

@@ -184,7 +184,7 @@ class VersionedTest extends SapphireTest {
 		$this->assertEquals($allPageIDs, $allPages->column('ID'));
 
 		// Check that this still works if we switch to reading the other stage
-		Versioned::reading_stage("Live");
+		Versioned::set_stage(Versioned::LIVE);
 		$allPages = Versioned::get_including_deleted("VersionedTest_DataObject", "\"ParentID\" = 0",
 			"\"VersionedTest_DataObject\".\"ID\" ASC");
 		$this->assertEquals(array("Page 1", "Page 2", "Page 3", "Subclass Page 1"), $allPages->column('Title'));
@@ -312,9 +312,9 @@ class VersionedTest extends SapphireTest {
 	}
 
 	public function testWritingNewToStage() {
-		$origStage = Versioned::current_stage();
+		$origStage = Versioned::get_stage();
 
-		Versioned::reading_stage("Stage");
+		Versioned::set_stage(Versioned::DRAFT);
 		$page = new VersionedTest_DataObject();
 		$page->Title = "testWritingNewToStage";
 		$page->URLSegment = "testWritingNewToStage";
@@ -331,7 +331,7 @@ class VersionedTest extends SapphireTest {
 		$this->assertEquals(1, $stage->count());
 		$this->assertEquals($stage->First()->Title, 'testWritingNewToStage');
 
-		Versioned::reading_stage($origStage);
+		Versioned::set_stage($origStage);
 	}
 
 	/**
@@ -341,9 +341,9 @@ class VersionedTest extends SapphireTest {
 	 * the VersionedTest_DataObject record though.
 	 */
 	public function testWritingNewToLive() {
-		$origStage = Versioned::current_stage();
+		$origStage = Versioned::get_stage();
 
-		Versioned::reading_stage("Live");
+		Versioned::set_stage(Versioned::LIVE);
 		$page = new VersionedTest_DataObject();
 		$page->Title = "testWritingNewToLive";
 		$page->URLSegment = "testWritingNewToLive";
@@ -360,7 +360,7 @@ class VersionedTest extends SapphireTest {
 		));
 		$this->assertEquals(0, $stage->count());
 
-		Versioned::reading_stage($origStage);
+		Versioned::set_stage($origStage);
 	}
 
 	/**
@@ -401,7 +401,7 @@ class VersionedTest extends SapphireTest {
 	 * Test that SQLSelect::queriedTables() applies the version-suffixes properly.
 	 */
 	public function testQueriedTables() {
-		Versioned::reading_stage('Live');
+		Versioned::set_stage(Versioned::LIVE);
 
 		$this->assertEquals(array(
 			'VersionedTest_DataObject_Live',
@@ -409,6 +409,102 @@ class VersionedTest extends SapphireTest {
 		), DataObject::get('VersionedTest_Subclass')->dataQuery()->query()->queriedTables());
 	}
 
+	/**
+	 * Virtual "sleep" that doesn't actually slow execution, only advances SS_DateTime::now()
+	 *
+	 * @param int $minutes
+	 */
+	protected function sleep($minutes) {
+		$now = SS_Datetime::now();
+		$date = DateTime::createFromFormat('Y-m-d H:i:s', $now->getValue());
+		$date->modify("+{$minutes} minutes");
+		SS_Datetime::set_mock_now($date->format('Y-m-d H:i:s'));
+	}
+
+	/**
+	 * Tests records selected by specific version
+	 */
+	public function testGetVersion() {
+		// Create a few initial versions to ensure this version
+		// doesn't clash with child versions
+		$this->sleep(1);
+		/** @var VersionedTest_DataObject $page2 */
+		$page2 = $this->objFromFixture('VersionedTest_DataObject', 'page2');
+		$page2->Title = 'dummy1';
+		$page2->write();
+		$this->sleep(1);
+		$page2->Title = 'dummy2';
+		$page2->write();
+		$this->sleep(1);
+		$page2->Title = 'Page 2 - v1';
+		$page2->write();
+		$version1Date = $page2->LastEdited;
+		$version1 = $page2->Version;
+
+		// Create another version where this object and some
+		// child records have been modified
+		$this->sleep(1);
+		/** @var VersionedTest_DataObject $page2a */
+		$page2a = $this->objFromFixture('VersionedTest_DataObject', 'page2a');
+		$page2a->Title = 'Page 2a - v2';
+		$page2a->write();
+		$this->sleep(1);
+		$page2->Title = 'Page 2 - v2';
+		$page2->write();
+		$version2Date = $page2->LastEdited;
+		$version2 = $page2->Version;
+		$this->assertGreaterThan($version1, $version2);
+		$this->assertDOSEquals(
+			[
+				['Title' => 'Page 2a - v2'],
+				['Title' => 'Page 2b'],
+			],
+			$page2->Children()
+		);
+
+		// test selecting v1
+		/** @var VersionedTest_DataObject $page2v1 */
+		$page2v1 = Versioned::get_version('VersionedTest_DataObject', $page2->ID, $version1);
+		$this->assertEquals('Page 2 - v1', $page2v1->Title);
+
+		// When selecting v1, related records should by filtered by
+		// the modified date of that version
+		$archiveParms = [
+			'Versioned.mode' => 'archive',
+			'Versioned.date' => $version1Date
+		];
+		$this->assertEquals($archiveParms, $page2v1->getInheritableQueryParams());
+		$this->assertArraySubset($archiveParms, $page2v1->Children()->getQueryParams());
+		$this->assertDOSEquals(
+			[
+				['Title' => 'Page 2a'],
+				['Title' => 'Page 2b'],
+			],
+			$page2v1->Children()
+		);
+
+		// When selecting v2, we get the same as on stage
+		/** @var VersionedTest_DataObject $page2v2 */
+		$page2v2 = Versioned::get_version('VersionedTest_DataObject', $page2->ID, $version2);
+		$this->assertEquals('Page 2 - v2', $page2v2->Title);
+
+		// When selecting v2, related records should by filtered by
+		// the modified date of that version
+		$archiveParms = [
+			'Versioned.mode' => 'archive',
+			'Versioned.date' => $version2Date
+		];
+		$this->assertEquals($archiveParms, $page2v2->getInheritableQueryParams());
+		$this->assertArraySubset($archiveParms, $page2v2->Children()->getQueryParams());
+		$this->assertDOSEquals(
+			[
+				['Title' => 'Page 2a - v2'],
+				['Title' => 'Page 2b'],
+			],
+			$page2v2->Children()
+		);
+	}
+
 	public function testGetVersionWhenClassnameChanged() {
 		$obj = new VersionedTest_DataObject;
 		$obj->Name = "test";
@@ -593,7 +689,7 @@ class VersionedTest extends SapphireTest {
 			'Does not contain separate table with _Stage suffix'
 		);
 
-		Versioned::reading_stage("Stage");
+		Versioned::set_stage(Versioned::DRAFT);
 		$obj = new VersionedTest_SingleStage(array('Name' => 'MyObj'));
 		$obj->write();
 		$this->assertNotNull(
@@ -601,7 +697,7 @@ class VersionedTest extends SapphireTest {
 			'Writes to and reads from default stage if its set explicitly'
 		);
 
-		Versioned::reading_stage("Live");
+		Versioned::set_stage(Versioned::LIVE);
 		$obj = new VersionedTest_SingleStage(array('Name' => 'MyObj'));
 		$obj->write();
 		$this->assertNotNull(
@@ -617,14 +713,14 @@ class VersionedTest extends SapphireTest {
 		$originalMode = Versioned::get_reading_mode();
 
 		// Generate staging record and retrieve it from stage in live mode
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 		$obj = new VersionedTest_Subclass();
 		$obj->Name = 'bob';
 		$obj->ExtraField = 'Field Value';
 		$obj->write();
 		$objID = $obj->ID;
 		$filter = sprintf('"VersionedTest_DataObject"."ID" = \'%d\'', Convert::raw2sql($objID));
-		Versioned::reading_stage('Live');
+		Versioned::set_stage(Versioned::LIVE);
 
 		// Check fields are unloaded prior to access
 		$objLazy = Versioned::get_one_by_stage('VersionedTest_DataObject', 'Stage', $filter, false);
@@ -831,7 +927,7 @@ class VersionedTest extends SapphireTest {
 
 		// Test that all (and only) public pages are viewable in stage mode
 		Session::clear("loggedInAs");
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 		$public1 = Versioned::get_one_by_stage('VersionedTest_PublicStage', 'Stage', array('"ID"' => $public1ID));
 		$public2 = Versioned::get_one_by_stage('VersionedTest_PublicViaExtension', 'Stage', array('"ID"' => $public2ID));
 		$private = Versioned::get_one_by_stage('VersionedTest_DataObject', 'Stage', array('"ID"' => $privateID));
@@ -841,7 +937,7 @@ class VersionedTest extends SapphireTest {
 		$this->assertFalse($private->canView());
 
 		// Adjusting the current stage should not allow objects loaded in stage to be viewable
-		Versioned::reading_stage('Live');
+		Versioned::set_stage(Versioned::LIVE);
 		$this->assertTrue($public1->canView());
 		$this->assertTrue($public2->canView());
 		$this->assertFalse($private->canView());
@@ -853,31 +949,30 @@ class VersionedTest extends SapphireTest {
 		$this->assertTrue($privateLive->canView());
 
 		// But if the private version becomes different to the live version, it's once again disallowed
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 		$private->Title = 'Secret Title';
 		$private->write();
 		$this->assertFalse($private->canView());
 		$this->assertTrue($privateLive->canView());
 
 		// And likewise, viewing a live page (when mode is draft) should be ok
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 		$this->assertFalse($private->canView());
 		$this->assertTrue($privateLive->canView());
 
 		// Logging in as admin should allow all permissions
 		$this->logInWithPermission('ADMIN');
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 		$this->assertTrue($public1->canView());
 		$this->assertTrue($public2->canView());
 		$this->assertTrue($private->canView());
 	}
 
-
 	public function testCanViewStage() {
 		$public = $this->objFromFixture('VersionedTest_PublicStage', 'public1');
 		$private = $this->objFromFixture('VersionedTest_DataObject', 'page1');
 		Session::clear("loggedInAs");
-		Versioned::reading_stage('Stage');
+		Versioned::set_stage(Versioned::DRAFT);
 
 		// Test that all (and only) public pages are viewable in stage mode
 		// Unpublished records are not viewable in live regardless of permissions
@@ -909,6 +1004,10 @@ class VersionedTest extends SapphireTest {
 
 
 /**
+ * @method VersionedTest_DataObject Parent()
+ * @method HasManyList Children()
+ * @method ManyManyList Related()
+ *
  * @package framework
  * @subpackage tests
  */
@@ -916,19 +1015,23 @@ class VersionedTest_DataObject extends DataObject implements TestOnly {
 	private static $db = array(
 		"Name" => "Varchar",
 		'Title' => 'Varchar',
-		'Content' => 'HTMLText'
+		'Content' => 'HTMLText',
 	);
 
 	private static $extensions = array(
-		"Versioned('Stage', 'Live')"
+		"Versioned('Stage', 'Live')",
 	);
 
 	private static $has_one = array(
-		'Parent' => 'VersionedTest_DataObject'
+		'Parent' => 'VersionedTest_DataObject',
+	);
+
+	private static $has_many = array(
+		'Children' => 'VersionedTest_DataObject',
 	);
 
 	private static $many_many = array(
-		'Related' => 'VersionedTest_RelatedWithoutVersion'
+		'Related' => 'VersionedTest_RelatedWithoutVersion',
 	);
 
 
@@ -948,7 +1051,7 @@ class VersionedTest_WithIndexes extends DataObject implements TestOnly {
 		'UniqS' => 'Int',
 	);
 	private static $extensions = array(
-		"Versioned('Stage', 'Live')"
+		"Versioned"
 	);
 	private static $indexes = array(
 		'UniqS_idx' => 'unique ("UniqS")',
@@ -1007,7 +1110,7 @@ class VersionedTest_SingleStage extends DataObject implements TestOnly {
 	);
 
 	private static $extensions = array(
-		'Versioned("Stage")'
+		'Versioned("Versioned")'
 	);
 }
 
@@ -1020,7 +1123,7 @@ class VersionedTest_PublicStage extends DataObject implements TestOnly {
 	);
 
 	private static $extensions = array(
-		"Versioned('Stage', 'Live')"
+		"Versioned"
 	);
 
 	public function canView($member = null) {
@@ -1055,7 +1158,7 @@ class VersionedTest_PublicViaExtension extends DataObject implements TestOnly {
 	);
 
 	private static $extensions = array(
-		"Versioned('Stage', 'Live')",
+		"Versioned",
 		"VersionedTest_PublicExtension"
 	);
 }

tests/view/SSViewerCacheBlockTest.php

@@ -145,11 +145,11 @@ class SSViewerCacheBlockTest extends SapphireTest {
 
 	public function testVersionedCache() {
 
-		$origStage = Versioned::current_stage();
+		$origStage = Versioned::get_stage();
 
 		// Run without caching in stage to prove data is uncached
 		$this->_reset(false);
-		Versioned::reading_stage("Stage");
+		Versioned::set_stage(Versioned::DRAFT);
 		$data = new SSViewerCacheBlockTest_VersionedModel();
 		$data->setEntropy('default');
 		$this->assertEquals(
@@ -165,7 +165,7 @@ class SSViewerCacheBlockTest extends SapphireTest {
 
 		// Run without caching in live to prove data is uncached
 		$this->_reset(false);
-		Versioned::reading_stage("Live");
+		Versioned::set_stage(Versioned::LIVE);
 		$data = new SSViewerCacheBlockTest_VersionedModel();
 		$data->setEntropy('default');
 		$this->assertEquals(
@@ -183,7 +183,7 @@ class SSViewerCacheBlockTest extends SapphireTest {
 		// changing the versioned reading mode doesn't cache between modes, but it does
 		// within them
 		$this->_reset(true);
-		Versioned::reading_stage("Stage");
+		Versioned::set_stage(Versioned::DRAFT);
 		$data = new SSViewerCacheBlockTest_VersionedModel();
 		$data->setEntropy('default');
 		$this->assertEquals(
@@ -197,7 +197,7 @@ class SSViewerCacheBlockTest extends SapphireTest {
 			$this->_runtemplate('<% cached %>$Inspect<% end_cached %>', $data)
 		);
 
-		Versioned::reading_stage('Live');
+		Versioned::set_stage(Versioned::LIVE);
 		$data = new SSViewerCacheBlockTest_VersionedModel();
 		$data->setEntropy('first');
 		$this->assertEquals(
@@ -211,7 +211,7 @@ class SSViewerCacheBlockTest extends SapphireTest {
 			$this->_runtemplate('<% cached %>$Inspect<% end_cached %>', $data)
 		);
 
-		Versioned::reading_stage($origStage);
+		Versioned::set_stage($origStage);
 	}
 
 	/**