State store

Overview

The state store enables non-persistent states to be stored in a distributed architecture.The distributed cache must be fast, scalable (several replicas) and built for high availability.

The state store stores data in key/value format, with the key indexed.

Consistency

The state store offers a choice of two consistency:

Strong consistency
Each node in the cluster responds with the latest data, even if the system has to block the query until all replicas have been updated. If you query a "consistent system" for a currently updated item, you won't get a response until all replicas have been properly updated. However, you will always receive the most up-to-date data.
Availability
Each node returns an immediate response, even if this response does not include the most recent data. If you query an "available system" for an item that's being updated, you'll get the best possible answer the service can provide at the time.

Concurrent access

The state store offers a choice between two types of optimistic concurrent access:

Last write wins
There is no concurrent access check, the key/value is updated without checking that there has been a competitive modification.
First write win
Use of ETAG to manage competitive access, if writing requires reloading the current data and modifying it.

Transactions

Neos does not manage transactions involving the writing of multiple keys/values.

Components

The neos run command uses the Redis component to implement the state store. Neos also recommends using Redis in production.

Note

The state store is available in development mode without Dapr (neos run -dp false command). In this case, the state store is stored as JSON files in the ~/.neos/statestore directory.

Architecture

graph TD
  CTA("Cluster Neos A") --> |"Update state by key"|STORE("Store (e.g. REDIS)"):::store
  STORE --> |"Get state by key"|CTB("Cluster Neos B")
  STORE --> |"Query State"|CTB
  CTC("Cluster Neos C") --> |"Update state by key"|STORE
  CTD("Cluster Neos D") --> |"Get state by key"|STORE
  classDef store fill:#C00306,color:#fff;

Getting started

In development mode, the state store is only available with the -dp option at startup, using the neos run -dp command.

To check that the necessary prerequisites (dapr, wsl, redis) have been installed, run the neos setup command first.

If you encounter problems with the neos setup command, you can manually install dapr and redis by following the documentation Install prerequisites manually.

For example, under Windows, the neos-state-redis.yaml file is created in C:\Users\[windows_user]\.neos\dapr\components.

Use of Neos State Store API

Neos expose the interface GroupeIsa.Neos.ClusterCommunication.DistributedStore.IStateStore to manage the state store.

To illustrate the use of the state store, we'll cache countries from an OpenData API call. The key will be the country's ISO code.

Add entry to the state store

To add an entry to the state store, you should use the method IStateStore.SaveStateAsync

1) Inject IStateStore

private readonly IHttpClientFactory _httpClientFactory;
private readonly IStateStore _countryStateStore;
private const string _keyPrefix = "Country/";

/// <summary>
/// Initializes a new instance of the <see cref="UpdateExternalCountryCache"/> class.
/// </summary>
/// <param name="httpClientFactory">The httpClientFactory.</param>
/// <param name="countryStateStore">The country state store.</param>
public UpdateExternalCountryCache(IHttpClientFactory httpClientFactory, IStateStore countryStateStore)
{
    _httpClientFactory = httpClientFactory;
    _countryStateStore = countryStateStore;
}

Note

The best practice is to generate a key with a prefix identifying the type of data (e.g. country), which allows you to use the state store for other data.
This choice is preferable if you wish to obtain information on a country quickly, as the state store does not allow you to obtain the list of keys, making it difficult to update or delete all the countries.

2) Examples of how to add or update a key

Update with concurrent access LastWrite

Example: Update a country in the case of a country cache.

await _stateStore.SaveStateAsync("FR", country, cancellationToken: cancellationToken);

Example: Update cache for all countries

await _countryStateStore.SaveStateAsync("Countries", countries);

Note

The default concurrent access mode is LastWrite and consistency is Strong.

Update with concurrent access FirstWrite

 bool succeded = await _countryStateStore.TrySaveStateAsync($"{_keyPrefix}/FR", country , etag ?? string.Empty, StateStoreConsistency.Availability, StateStoreConcurrency.FirstWrite, CancellationToken.None);

Note

To create a new key/value, etag must be set to string.Empty; to update, etag must first be retrieved using the IStateStore.GetStateAsync method.

3) Examples of how to add a key with a time to live

await _countryStateStore.SaveStateAsync("{_keyPrefix}/FR", country, new StateStoreSaveOptions().WithTtlInSeconds(3600), cancellationToken: cancellationToken);

In this example, the key will be deleted after 3600 seconds.

Note

The time to live option is available in the state store in development mode without Dapr, however ttl is only available during the lifetime of the application. So the file in the ~/.neos/statestore directory will not be deleted if the application is restarted before the ttl expires.

Get entry from the state store

Get a key/value from the store by its key :

(Country country, string etag) =  await _countryStateStore.GetStateAsync($"{_keyPrefix}/FR", cancellationToken: cancellationToken);

Delete entry from the state store

Delete a key/value from the store by its key :

await _countryStateStore.DeleteStateAsync($"{_keyPrefix}/FR"), cancellationToken: cancellationToken;

Bulk operations

Create a key/value set :

ExternalCountries countries = await GetExternalCountriesAsync();
await _countryStateStore.SaveBulkStateAsync<ExternalCountry>(countries.Records.Select(c => new BulkStateStoreItem<ExternalCountry>($"{_keyPrefix}/c.Fields.Iso2", c.Fields, string.Empty), cancellationToken).ToList());

In this example, we obtain data from an OpenData API and then cache it in bulk mode using the SaveBulkStateAsync method.

Delete a key/value set :

(CachedCountry[]? cachedCountries, string eTag) = await _countryStateStore.GetStateAsync<CachedCountry[]>("Countries");
if (cachedCountries != null)
{
  await _countryStateStore.DeleteBulkStateAsync(
      cachedCountries.Values
      .Where(c => c.Key.StartsWith(_keyPrefix))
      .Select(c => new BulkStateStoreItem(c.Key, c.ETag)).ToList(),
      cancellationToken: cancellationToken);
}

In this example, we obtain all the keys/values, then delete the country keys/values.

The Framework's use of the StateStore

The Framework uses the StateStore internally for the following purposes:

Name	Key	Description
Chatbot threads	`ChatbotThread_{ThreadId}`	Stores threads for the chatbot.
Server data exports	`ExportStatus_{ExportId}`	Manages the status of data exports.
Cluster versions by tenant	`Tenants_ClusterVersionsByTenant_({TenantIdentifier})`	Maintains a list of cluster versions associated with each tenant.
Resolution by tenant	`Tenants_ResolutionByTenant_({TenantIdentifierLowercase})`	Keeps track of resolution information for tenants.
Clusters by user	`Tenants_ClustersByUser_({UserLoginLowercase})`	Stores information on clusters associated with users.
User permissions	`{ClusterName}_{ClusterVersion}_{TenantId}_UserPermissions_{UserId}`	Handles user permissions.
Referenced modules	`{ClusterName}_{ClusterVersion}_Modules`	The list of module versions referenced in a cluster version.
Deployment configuration	`NeosDeploymentConfiguration`	The configuration of deployed services.

Under the hood

GroupeIsa.Neos.ClusterCommunication.DistributedStore.IStateStore use the building block DAPR StateStore.

Manually install prerequisites

Windows

Install DAPR (mode self-hosted)

winget install Dapr.CLI -V 1.10.0

Enable WSL2 and install a distribution

Install WSL2. Recommended distributions are Ubuntu or Debian.

Install Redis To install Redis follow the instructions at Redis install

Linux (Ubuntu, Debian)

Install DAPR (mode self-hosted)

wget -q https://raw.githubusercontent.com/dapr/cli/master/install/install.sh -O - | /bin/bash -s 1.10.0

Install Redis To install Redis follow the instructions at Redis install

Uninstall Redis

To uninstall Redis, use the following command :

sudo apt-get purge --auto-remove redis-server

Viewing the Redis cache

Use redis-cli

Under Linux or WSL for Windows, launch the Redi client with the redis-cli command.

To view keys: Command KEY

Example for viewing all keys :
```
KEY *
```
To view the JSON value of a key : Command JSON.GET
```
JSON.GET Country/FR
```
Delete a key: Command DEL
```
DEL Country/FR
```
View index list Command FT._LIST
```
FT._LIST
```
Delete an index Command FT.DROPINDEX
```
FT.DROPINDEX europeanUnionIdx
```
Delete all keys command FLUSHDB
```
FLUSHDB
```

Warning

If the Redis instance is also used for pub/sub, all services using it must be restarted after this command.

Extension VS Code

You can use Visual Studio Code Extension like vscode-redis-client to visualize keys, delete keys, etc ...

To view the Redis cache of a Kubernetes pod, you can perform a PORT-FORWARD :

kubectl port-forward [name-of-redis-pod] 6379 6380

And connect to the redis server 127.0.0.1:6380

Table of Contents

State store

Overview

Consistency

Concurrent access

Transactions

Components

Note

Architecture

Getting started

Use of Neos State Store API

Add entry to the state store

Note

Note

Note

Note

Get entry from the state store

Delete entry from the state store

Bulk operations

The Framework's use of the StateStore

Under the hood

Manually install prerequisites

Windows

Linux (Ubuntu, Debian)

Uninstall Redis

Viewing the Redis cache

Use redis-cli

Warning

Extension VS Code